package records

// Rejected-save audit coverage (issue #4261).
//
// The obligations this file discharges:
//   - one case per rejection stage reaching a real write entry point
//     (field validation / validation rule / uniqueness / FLS) across
//     insert, update, and bulk;
//   - the best-effort mutation test: force the sink to fail (and to panic)
//     and assert the user-visible request is unaffected;
//   - the PII red line: field names are recorded, values never are;
//   - no recursion: emitting a failure row produces no further audit events.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"

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

type auditNeutralValidationError struct {
	field   string
	message string
}

func (e auditNeutralValidationError) Error() string           { return e.message }
func (e auditNeutralValidationError) ValidationField() string { return e.field }
func (e auditNeutralValidationError) ValidationMessage() string {
	panic("rejected-save audit must not read validation messages")
}

type auditNeutralWriteDeniedError struct {
	fields  []string
	message string
}

func (e auditNeutralWriteDeniedError) Error() string { return e.message }
func (e auditNeutralWriteDeniedError) FieldWriteDeniedFields() []string {
	return append([]string(nil), e.fields...)
}
func (e auditNeutralWriteDeniedError) FieldWriteDeniedMessage() string {
	panic("rejected-save audit must not read field-denial messages")
}

// --- test doubles ---------------------------------------------------------

// captureFailureSink records every emitted event and can be made to fail.
type captureFailureSink struct {
	events   []audit.AuditEvent
	failWith error
	panicOn  bool
	calls    int
}

func (s *captureFailureSink) EmitFailure(_ context.Context, event audit.AuditEvent) error {
	s.calls++
	if s.panicOn {
		panic("audit sink exploded")
	}
	if s.failWith != nil {
		return s.failWith
	}
	s.events = append(s.events, event)
	return nil
}

func (s *captureFailureSink) only(t *testing.T) audit.AuditEvent {
	t.Helper()
	if len(s.events) != 1 {
		t.Fatalf("emitted %d rejected-save rows, want exactly 1: %+v", len(s.events), s.events)
	}
	return s.events[0]
}

func decodeChanges(t *testing.T, event audit.AuditEvent) rejectedSaveChanges {
	t.Helper()
	var changes rejectedSaveChanges
	if err := json.Unmarshal([]byte(event.ChangesJSON), &changes); err != nil {
		t.Fatalf("ChangesJSON %q is not valid JSON: %v", event.ChangesJSON, err)
	}
	return changes
}

func newAuditedEntrypointService(schema *models.ObjectMetadata, repo *captureRecordRepo) (*PersistenceService, *captureFailureSink) {
	ps := newRecordEntrypointTestService(schema, repo)
	sink := &captureFailureSink{}
	ps.SetRejectedSaveAuditSink(sink)
	return ps, sink
}

// --- classifier -----------------------------------------------------------

func TestClassifyRejectedSave(t *testing.T) {
	for _, tc := range []struct {
		name      string
		err       error
		wantOK    bool
		wantStage string
		wantRule  string
		wantField string
		wantCode  string
	}{
		{
			name:      "required field",
			err:       appErrors.NewValidationError("Stage__c", "is required"),
			wantOK:    true,
			wantStage: rejectionStageFieldValidation,
			wantField: "Stage__c",
		},
		{
			name:      "validation rule",
			err:       appErrors.NewValidationRuleError("Block_Everything", "blocked"),
			wantOK:    true,
			wantStage: rejectionStageValidationRule,
			wantRule:  "Block_Everything",
		},
		{
			name: "aggregated validation rules",
			err: appErrors.NewValidationRuleFailures([]appErrors.ValidationRuleFailure{
				{RuleName: "IQM_Existing_Database", Message: "Existing Database is required", Field: "Existing_Database__c"},
				{RuleName: "IQM_Completed_Date", Message: "IQM Completed Date is required", Field: "IQM_Completed_Date__c"},
			}),
			wantOK:    true,
			wantStage: rejectionStageValidationRule,
			wantRule:  "IQM_Existing_Database",
			wantField: "Existing_Database__c",
		},
		{
			name:      "fls denial",
			err:       appErrors.NewFLSDeniedErrorForFields([]string{"Secret__c"}),
			wantOK:    true,
			wantStage: rejectionStageFLS,
			wantField: "Secret__c",
		},
		{
			name:      "neutral field validation",
			err:       auditNeutralValidationError{field: "Neutral_Field__c", message: "is invalid"},
			wantOK:    true,
			wantStage: rejectionStageFieldValidation,
			wantField: "Neutral_Field__c",
			wantCode:  string(sharedErrors.ErrValidation),
		},
		{
			name:      "invalid write field remains an audited validation rejection",
			err:       appErrors.NewInvalidFieldForWriteError([]string{"CatalogNamespaceId"}, "field is catalog-owned"),
			wantOK:    true,
			wantStage: rejectionStageFieldValidation,
			wantField: "CatalogNamespaceId",
			wantCode:  string(sharedErrors.ErrValidation),
		},
		{
			name:      "neutral field write denial",
			err:       auditNeutralWriteDeniedError{fields: []string{"Neutral_Secret__c"}, message: "field is not writable"},
			wantOK:    true,
			wantStage: rejectionStageFLS,
			wantField: "Neutral_Secret__c",
			wantCode:  string(sharedErrors.ErrValidation),
		},
		{
			name:      "uniqueness conflict",
			err:       appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1"),
			wantOK:    true,
			wantStage: rejectionStageUniqueness,
			wantField: "ExternalId__c",
		},
		{
			// The single commonest write rejection there is — an OwnerId or
			// lookup pointing at a record that no longer exists — arrives as
			// DomainWireError, which is NOT one of the four typed errors the
			// #4261 handoff listed. Classifying it is what keeps the surface
			// from looking complete while staying dark on the common case.
			name:      "invalid reference (domain wire error)",
			err:       sharedErrors.NewInvalidReferenceError("OwnerId: referenced User record \"005-ghost\" does not exist", constants.FieldOwnerID),
			wantOK:    true,
			wantStage: rejectionStageDomainRule,
			wantField: constants.FieldOwnerID,
		},
		{
			name:      "lifecycle refusal (apierr 4xx)",
			err:       apierr.Invalid("ParentRoleId would create a cycle in role hierarchy"),
			wantOK:    true,
			wantStage: rejectionStageLifecycle,
		},
		{
			// A 5xx apierr is infrastructure, not a refused save — same
			// reasoning as the IsInternal guard.
			name:   "apierr 5xx is not a rejection",
			err:    apierr.Internal(errors.New("disk on fire")),
			wantOK: false,
		},
		{
			name:   "nil error",
			err:    nil,
			wantOK: false,
		},
		{
			name:   "untyped error is not a rejection",
			err:    errors.New("boom"),
			wantOK: false,
		},
		{
			// The transaction manager wraps rollback failures as
			// InternalError(cause=businessErr). Without the IsInternal guard,
			// errors.As would unwrap into the business cause and record a
			// genuine infrastructure failure as a user validation rejection.
			name:   "internal error wrapping a validation error",
			err:    appErrors.NewInternalError("rollback failed", appErrors.NewValidationError("Stage__c", "is required")),
			wantOK: false,
		},
		{
			// Wrapping must still classify: rejections routinely reach the
			// entry point inside a %w chain.
			name:      "wrapped validation error",
			err:       fmt.Errorf("insert failed: %w", appErrors.NewValidationError("Name", "is required")),
			wantOK:    true,
			wantStage: rejectionStageFieldValidation,
			wantField: "Name",
		},
		{
			name: "aggregate classification follows first validation source",
			err: appErrors.NewClientValidationErrors(
				appErrors.NewValidationError("Name", "is required"),
				appErrors.NewValidationRuleError("Industry_Required", "Industry is required"),
			),
			wantOK:    true,
			wantStage: rejectionStageFieldValidation,
			wantField: "Name",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			got, ok := classifyRejectedSave(tc.err)
			if ok != tc.wantOK {
				t.Fatalf("classifyRejectedSave(%v) ok = %v, want %v", tc.err, ok, tc.wantOK)
			}
			if !tc.wantOK {
				return
			}
			if got.stage != tc.wantStage {
				t.Errorf("stage = %q, want %q", got.stage, tc.wantStage)
			}
			if got.rule != tc.wantRule {
				t.Errorf("rule = %q, want %q", got.rule, tc.wantRule)
			}
			if tc.wantField != "" {
				if len(got.fields) != 1 || got.fields[0] != tc.wantField {
					t.Errorf("fields = %v, want [%q]", got.fields, tc.wantField)
				}
			}
			if got.errorCode == "" {
				t.Error("errorCode is empty; the ErrorCode column would stay NULL")
			}
			if tc.wantCode != "" && got.errorCode != tc.wantCode {
				t.Errorf("errorCode = %q, want %q", got.errorCode, tc.wantCode)
			}
		})
	}
}

func TestRejectedSaveAudit_EmitsNeutralValidationConditions(t *testing.T) {
	for _, tc := range []struct {
		name      string
		err       error
		wantStage string
	}{
		{
			name:      "field validation",
			err:       auditNeutralValidationError{field: "Neutral_Field__c", message: "customer-authored message"},
			wantStage: rejectionStageFieldValidation,
		},
		{
			name:      "field write denial",
			err:       auditNeutralWriteDeniedError{fields: []string{"Neutral_Secret__c"}, message: "customer-authored message"},
			wantStage: rejectionStageFLS,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			sink := &captureFailureSink{}
			ps := &PersistenceService{rejectedSaveAuditSink: sink}
			ps.emitRejectedSaveAudit(
				context.Background(),
				"Account",
				"001-rec",
				audit.ActionUpdate,
				nil,
				tc.err,
				recordEntrypointUser(),
			)

			event := sink.only(t)
			if event.ErrorCode != string(sharedErrors.ErrValidation) {
				t.Fatalf("ErrorCode = %q, want %q", event.ErrorCode, sharedErrors.ErrValidation)
			}
			if changes := decodeChanges(t, event); changes.RejectionStage != tc.wantStage {
				t.Fatalf("rejectionStage = %q, want %q", changes.RejectionStage, tc.wantStage)
			}
		})
	}
}

// --- PII red line ---------------------------------------------------------

// TestRejectedSaveAudit_NeverRecordsValues is the PII guardrail.
//
// Every typed error here carries a value in its rendered message
// (ConflictError formats the conflicting value; the picklist validator
// formats the offending value; a validation rule's admin-authored message can
// render merge fields). The audit row must contain none of them.
func TestRejectedSaveAudit_NeverRecordsValues(t *testing.T) {
	const secretValue = "SUPER-SECRET-CUSTOMER-VALUE"

	for _, tc := range []struct {
		name string
		err  error
	}{
		{"conflict carries the value", appErrors.NewConflictError("Account", "ExternalId__c", secretValue)},
		{"picklist validator renders the value", appErrors.NewValidationError("Stage__c", fmt.Sprintf("value %q is not an active picklist option", secretValue))},
		{"validation rule message renders a merge field", appErrors.NewValidationRuleError("Amount_Rule", "Amount "+secretValue+" is too large")},
	} {
		t.Run(tc.name, func(t *testing.T) {
			// Sanity: the error itself really does leak the value, so this
			// test is exercising a live hazard and not a strawman.
			if !strings.Contains(tc.err.Error(), secretValue) {
				t.Fatalf("fixture error %q does not contain the value; test would pass vacuously", tc.err)
			}

			sink := &captureFailureSink{}
			ps := &PersistenceService{rejectedSaveAuditSink: sink}
			ps.emitRejectedSaveAudit(context.Background(), "Account", "001-rec", audit.ActionCreate,
				[]string{"ExternalId__c", "Stage__c"}, tc.err, recordEntrypointUser())

			event := sink.only(t)
			// Scan the whole row, not just the field we expect the value in:
			// a future field added to AuditEvent must not reintroduce a leak.
			serialized, err := json.Marshal(event)
			if err != nil {
				t.Fatalf("marshal event: %v", err)
			}
			if strings.Contains(string(serialized), secretValue) {
				t.Fatalf("audit row leaked a field VALUE: %s", serialized)
			}
		})
	}
}

// TestRejectedSaveAudit_DropsUndeclaredFieldNames is the OTHER half of the PII
// rule, and the half that is easy to miss: not every "field name" reaching
// this code is a field name.
//
// Two real paths feed caller-controlled strings into the name position:
//   - the unknown-field bouncer builds its ValidationError with the RAW
//     caller-supplied map key as .Field;
//   - the CSV bulk path turns file headers straight into map keys, so a
//     headerless or off-by-one upload makes the first DATA row the header and
//     every key becomes a customer value.
//
// Recording those verbatim would put customer data into a 78-month table
// through the very feature that promises not to. Only names the schema
// declares are recorded; the rest collapse to a count.
func TestRejectedSaveAudit_DropsUndeclaredFieldNames(t *testing.T) {
	const leakedValue = "jane.doe@acme.example"

	repo := &captureRecordRepo{}
	ps, sink := newAuditedEntrypointService(recordEntrypointSchema(), repo)

	// Shape of a mis-headered CSV row: one real field, two customer values
	// sitting where field names belong.
	ps.emitRejectedSaveAudit(context.Background(), "ProbeObject", "rec-1", audit.ActionCreate,
		[]string{constants.FieldName, leakedValue, "+1-415-555-0134"},
		appErrors.NewValidationError(leakedValue, "Field does not exist"),
		recordEntrypointUser())

	event := sink.only(t)
	serialized, err := json.Marshal(event)
	if err != nil {
		t.Fatalf("marshal event: %v", err)
	}
	if strings.Contains(string(serialized), leakedValue) {
		t.Fatalf("audit row persisted a caller-supplied string as if it were a field name: %s", serialized)
	}
	if strings.Contains(string(serialized), "555-0134") {
		t.Fatalf("audit row persisted a caller-supplied string as if it were a field name: %s", serialized)
	}

	changes := decodeChanges(t, event)
	// The declared field survives — dropping everything would be safe but
	// useless, and would hide a regression that broke the whole payload.
	if len(changes.RequestedFields) != 1 || changes.RequestedFields[0] != constants.FieldName {
		t.Fatalf("requestedFields = %v, want [%s]: declared fields must still be recorded",
			changes.RequestedFields, constants.FieldName)
	}
	// The forensic signal is preserved as a count, so support can still see
	// "the caller sent fields this object does not have".
	if changes.UnknownFieldCount == 0 {
		t.Fatal("unknownFieldCount = 0; dropped names must still be counted or the row hides that the caller sent unknown fields")
	}
}

// TestRejectedSaveAudit_NoSchemaRecordsNoNames pins the fail-safe direction of
// the filter: with metadata unavailable there is no way to tell a field name
// from a customer value, so nothing is recorded. Losing the field list is
// recoverable; leaking data into a long-retention table is not.
func TestRejectedSaveAudit_NoSchemaRecordsNoNames(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink} // no metadata service

	ps.emitRejectedSaveAudit(context.Background(), "ProbeObject", "rec-1", audit.ActionCreate,
		[]string{"Name", "secret-value"},
		appErrors.NewValidationError("secret-value", "Field does not exist"),
		recordEntrypointUser())

	changes := decodeChanges(t, sink.only(t))
	if len(changes.RequestedFields) != 0 || len(changes.Fields) != 0 {
		t.Fatalf("names recorded without a schema to validate them against: %+v", changes)
	}
	if changes.UnknownFieldCount == 0 {
		t.Fatal("unknownFieldCount = 0; the row must still say names were dropped")
	}
}

func TestRequestedFieldNames_RecordsNamesNotValues(t *testing.T) {
	names := requestedFieldNames(models.SObject{
		"Name":          "ACME Corporation",
		"Amount":        12345.67,
		"ExternalId__c": "SECRET-1",
	})
	want := []string{"Amount", "ExternalId__c", "Name"}
	if len(names) != len(want) {
		t.Fatalf("requestedFieldNames() = %v, want %v", names, want)
	}
	for i := range want {
		if names[i] != want[i] {
			t.Fatalf("requestedFieldNames() = %v, want %v (sorted for deterministic ChangesJSON)", names, want)
		}
	}
}

// --- best-effort contract (mutation tests) --------------------------------

// TestRejectedSaveAudit_SinkFailureDoesNotAffectRequest is the mutation test
// the #4261 handoff requires: break the audit write and assert the user's
// request is byte-for-byte unaffected.
//
// A guard alone would not prove this — the assertion has to be that the
// FAILING sink was actually reached (sink.calls > 0) and the caller's error
// still came back unchanged. An audit path that can turn a validation
// rejection into a 500 is worse than no audit path.
func TestRejectedSaveAudit_SinkFailureDoesNotAffectRequest(t *testing.T) {
	for _, tc := range []struct {
		name string
		sink *captureFailureSink
	}{
		{"sink returns an error", &captureFailureSink{failWith: errors.New("audit database is down")}},
		{"sink panics", &captureFailureSink{panicOn: true}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			repo := &captureRecordRepo{}
			healthy := &captureFailureSink{}
			schema := requiredPicklistRecordEntrypointSchema()

			// Baseline: the same rejection with a working sink. Whatever the
			// caller sees here is what a broken sink must not change.
			baselineService := newRecordEntrypointTestService(schema, repo)
			baselineService.SetRejectedSaveAuditSink(healthy)
			_, baselineErr := baselineService.Insert(context.Background(), "ProbeObject",
				models.SObject{constants.FieldName: "probe"}, recordEntrypointUser())
			if baselineErr == nil {
				t.Fatal("baseline Insert succeeded; the fixture no longer produces a rejection")
			}

			brokenRepo := &captureRecordRepo{}
			brokenService := newRecordEntrypointTestService(requiredPicklistRecordEntrypointSchema(), brokenRepo)
			brokenService.SetRejectedSaveAuditSink(tc.sink)

			result, err := brokenService.Insert(context.Background(), "ProbeObject",
				models.SObject{constants.FieldName: "probe"}, recordEntrypointUser())

			if tc.sink.calls == 0 {
				t.Fatal("the broken sink was never reached; this test would pass vacuously")
			}
			if err == nil {
				t.Fatal("Insert returned nil error after an audit failure; the audit path swallowed the rejection")
			}
			if err.Error() != baselineErr.Error() {
				t.Fatalf("audit failure changed the user-visible error:\n got: %v\nwant: %v", err, baselineErr)
			}
			if result != nil {
				t.Fatalf("Insert returned a record for a rejected save: %#v", result)
			}
			if brokenRepo.inserted != nil {
				t.Fatalf("rejected record was persisted: %#v", brokenRepo.inserted)
			}
		})
	}
}

// TestRejectedSaveAudit_SuccessfulSaveEmitsNothing guards the other direction:
// the hook must be inert on the happy path.
func TestRejectedSaveAudit_SuccessfulSaveEmitsNothing(t *testing.T) {
	repo := &captureRecordRepo{}
	ps, sink := newAuditedEntrypointService(recordEntrypointSchema(), repo)

	if _, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "probe"}, recordEntrypointUser()); err != nil {
		t.Fatalf("Insert() error = %v, want success", err)
	}
	if repo.inserted == nil {
		t.Fatal("Insert did not persist; fixture is not exercising the success path")
	}
	if sink.calls != 0 {
		t.Fatalf("a successful save emitted %d rejected-save rows, want 0", sink.calls)
	}
}

// --- per-path, per-stage coverage ----------------------------------------

func TestRejectedSaveAudit_InsertFieldValidation(t *testing.T) {
	repo := &captureRecordRepo{}
	ps, sink := newAuditedEntrypointService(requiredPicklistRecordEntrypointSchema(), repo)

	_, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "probe"}, recordEntrypointUser())
	if err == nil {
		t.Fatal("Insert accepted a record missing a required picklist")
	}

	event := sink.only(t)
	if event.Outcome != audit.OutcomeFailure {
		t.Errorf("Outcome = %q, want %q", event.Outcome, audit.OutcomeFailure)
	}
	if event.Action != audit.ActionCreate {
		t.Errorf("Action = %q, want %q", event.Action, audit.ActionCreate)
	}
	if event.ObjectApiName != "ProbeObject" {
		t.Errorf("ObjectApiName = %q, want ProbeObject", event.ObjectApiName)
	}
	if event.ErrorCode == "" {
		t.Error("ErrorCode is empty; #4261 exists because that column was always NULL")
	}

	changes := decodeChanges(t, event)
	if changes.RejectionStage != rejectionStageFieldValidation {
		t.Errorf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageFieldValidation)
	}
	if len(changes.Fields) != 1 || changes.Fields[0] != "Stage__c" {
		t.Errorf("fields = %v, want [Stage__c]", changes.Fields)
	}
	// The caller's payload set is the "what were they trying to do" evidence,
	// and the missing required field is precisely what is NOT in it.
	if len(changes.RequestedFields) != 1 || changes.RequestedFields[0] != constants.FieldName {
		t.Errorf("requestedFields = %v, want [%s]", changes.RequestedFields, constants.FieldName)
	}
	for _, name := range changes.RequestedFields {
		if name == "Stage__c" {
			t.Error("requestedFields contains the missing required field; a missing-required-field rejection would be indistinguishable from a bad-value one")
		}
	}
}

func TestRejectedSaveAudit_InsertUniqueness(t *testing.T) {
	schema := recordEntrypointSchema()
	for i := range schema.Fields {
		if schema.Fields[i].APIName == constants.FieldName {
			schema.Fields[i].IsUnique = true
		}
	}
	makeFieldsMap(schema)

	repo := &captureRecordRepo{
		stubUpsertRecordRepo: stubUpsertRecordRepo{
			checkUniqueFn: func(context.Context, *models.ObjectMetadata, string, interface{}, string) (bool, error) {
				return true, nil
			},
		},
	}
	ps, sink := newAuditedEntrypointService(schema, repo)

	_, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "duplicate-value"}, recordEntrypointUser())
	if err == nil {
		t.Fatal("Insert accepted a duplicate value on a unique field")
	}

	event := sink.only(t)
	changes := decodeChanges(t, event)
	if changes.RejectionStage != rejectionStageUniqueness {
		t.Fatalf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageUniqueness)
	}
	if len(changes.Fields) != 1 || changes.Fields[0] != constants.FieldName {
		t.Errorf("fields = %v, want [%s]", changes.Fields, constants.FieldName)
	}
	if strings.Contains(event.ChangesJSON, "duplicate-value") {
		t.Fatalf("uniqueness row leaked the conflicting value: %s", event.ChangesJSON)
	}
	// Uniqueness runs inside the transaction, so this row also proves the
	// out-of-band write survives the rollback that follows.
	if repo.inserted != nil {
		t.Fatalf("duplicate record was persisted: %#v", repo.inserted)
	}
}

func TestRejectedSaveAudit_UpdateRecordsTheRecordId(t *testing.T) {
	schema := requiredPicklistRecordEntrypointSchema()
	repo := &captureRecordRepo{
		old: models.SObject{
			constants.FieldID:   "rec-42",
			constants.FieldName: "existing",
			"Stage__c":          "Open",
		},
	}
	ps, sink := newAuditedEntrypointService(schema, repo)

	// Explicitly blanking a required picklist is a rejection on update.
	err := ps.Update(context.Background(), "ProbeObject", "rec-42",
		models.SObject{"Stage__c": ""}, recordEntrypointUser())
	if err == nil {
		t.Fatal("Update accepted a blanked required picklist")
	}

	event := sink.only(t)
	if event.Action != audit.ActionUpdate {
		t.Errorf("Action = %q, want %q", event.Action, audit.ActionUpdate)
	}
	if event.RecordId != "rec-42" {
		t.Errorf("RecordId = %q, want rec-42 — an update rejection knows exactly which record it refused", event.RecordId)
	}
	changes := decodeChanges(t, event)
	if changes.RejectionStage != rejectionStageFieldValidation {
		t.Errorf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageFieldValidation)
	}
	if len(changes.RequestedFields) != 1 || changes.RequestedFields[0] != "Stage__c" {
		t.Errorf("requestedFields = %v, want [Stage__c]", changes.RequestedFields)
	}
	if repo.updated != nil {
		t.Fatalf("rejected update was persisted: %#v", repo.updated)
	}
}

func TestRejectedSaveAudit_UpdateFLSDenial(t *testing.T) {
	schema := recordEntrypointSchema()
	schema.Fields = append(schema.Fields, models.FieldMetadata{
		APIName:    "Secret__c",
		Type:       constants.FieldTypeString,
		Createable: true,
		Updateable: true,
	})
	makeFieldsMap(schema)

	repo := &captureRecordRepo{
		old: models.SObject{
			constants.FieldID:   "rec-7",
			constants.FieldName: "existing",
			"Secret__c":         "old",
		},
	}
	ps, sink := newAuditedEntrypointService(schema, repo)
	ps.permissions = permissions.NewPermissionService(&updateFLSBoundedPermRepo{
		stubUpsertPermRepo: stubUpsertPermRepo{grantCreate: true, grantEdit: true},
		fieldPerm: &models.FieldPermissions{
			SobjectType:     "ProbeObject",
			Field:           "Secret__c",
			PermissionsRead: true,
			PermissionsEdit: false,
		},
	}, ps.metadata, nil)

	err := ps.Update(context.Background(), "ProbeObject", "rec-7",
		models.SObject{"Secret__c": "new-secret"}, recordEntrypointUser())
	if err == nil {
		t.Fatal("Update accepted a write to an FLS-denied field")
	}
	if _, ok := appErrors.AsFLSDenied(err); !ok {
		t.Fatalf("Update error = %T %[1]v, want FLSDeniedError; fixture is not exercising the FLS path", err)
	}

	event := sink.only(t)
	changes := decodeChanges(t, event)
	if changes.RejectionStage != rejectionStageFLS {
		t.Fatalf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageFLS)
	}
	if len(changes.Fields) != 1 || changes.Fields[0] != "Secret__c" {
		t.Errorf("fields = %v, want [Secret__c] — FLS reports every denied field", changes.Fields)
	}
	if strings.Contains(event.ChangesJSON, "new-secret") {
		t.Fatalf("FLS row leaked the denied value: %s", event.ChangesJSON)
	}
}

func TestRejectedSaveAudit_InsertValidationRule(t *testing.T) {
	schema := recordEntrypointSchema()
	repo := &captureRecordRepo{}
	ps, sink := newAuditedEntrypointService(schema, repo)

	// Rebuild the metadata service on a repo that publishes an always-firing
	// rule; the shared entrypoint constructor hardcodes a rule-less one.
	rules := []*models.ValidationRule{{
		ID:            "vr-block",
		ObjectAPIName: "ProbeObject",
		Name:          "Block_Everything",
		Active:        true,
		Condition:     "true",
		ErrorMessage:  "blocked by test rule",
	}}
	ps.metadata = metadataFeature.NewMetadataService(
		&stubUpsertMetaRepo{schema: schema, validationRules: rules}, nil, nil)
	ps.permissions = permissions.NewPermissionService(
		&stubUpsertPermRepo{grantCreate: true, grantEdit: true}, ps.metadata, nil)

	_, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "probe"}, recordEntrypointUser())
	if err == nil {
		t.Fatal("Insert accepted a record blocked by an active validation rule")
	}

	event := sink.only(t)
	changes := decodeChanges(t, event)
	if changes.RejectionStage != rejectionStageValidationRule {
		t.Fatalf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageValidationRule)
	}
	if changes.Rule != "Block_Everything" {
		t.Errorf("rule = %q, want Block_Everything — the rule name is the whole point of this stage", changes.Rule)
	}
	// The admin-authored ErrorMessage is never recorded: it can render merge
	// fields, i.e. record values.
	if strings.Contains(event.ChangesJSON, "blocked by test rule") {
		t.Fatalf("row recorded the rule's rendered message, which can carry values: %s", event.ChangesJSON)
	}
}

// TestRejectedSaveAudit_RecycleBinRestoreUniqueness covers the fourth write
// path. Restore re-runs reactivation constraints and the same uniqueness
// check as a normal save, so it can be refused for the same reasons — and
// before #4261 it left the same silence behind.
func TestRejectedSaveAudit_RecycleBinRestoreUniqueness(t *testing.T) {
	user := recordEntrypointUser()
	schema := restoreObjectSchema("Account")
	for i := range schema.Fields {
		if schema.Fields[i].APIName == constants.FieldName {
			schema.Fields[i].IsUnique = true
		}
	}
	makeFieldsMap(schema)

	repo := newRecycleBinRestoreRepo("Account", "rec-1", user.ID)
	// A LIVE row already holds the deleted row's unique Name, so restoring
	// would collide.
	repo.records["account|rec-live"] = models.SObject{
		constants.FieldID:        "rec-live",
		constants.FieldName:      "Deleted row",
		constants.FieldOwnerID:   user.ID,
		constants.FieldIsDeleted: false,
	}

	ps := newRecycleBinRestoreService(t, repo, true, schema)
	sink := &captureFailureSink{}
	ps.SetRejectedSaveAuditSink(sink)

	err := ps.RestoreRecord(context.Background(), "rec-1", user)
	if err == nil {
		t.Fatal("RestoreRecord accepted a restore that collides on a unique field")
	}

	event := sink.only(t)
	if event.Action != audit.ActionUndelete {
		t.Errorf("Action = %q, want %q — a restore attempts UNDELETE", event.Action, audit.ActionUndelete)
	}
	if event.ObjectApiName != "Account" {
		t.Errorf("ObjectApiName = %q, want Account", event.ObjectApiName)
	}
	if event.RecordId != "rec-1" {
		t.Errorf("RecordId = %q, want rec-1", event.RecordId)
	}
	changes := decodeChanges(t, event)
	if changes.RejectionStage != rejectionStageUniqueness {
		t.Fatalf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageUniqueness)
	}
	// A restore carries no caller payload — the user supplied a record id,
	// not fields — so nothing is claimed about their intent.
	if len(changes.RequestedFields) != 0 {
		t.Errorf("requestedFields = %v, want empty for a restore", changes.RequestedFields)
	}
}

// --- gating ---------------------------------------------------------------

func TestRejectedSaveAudit_NoSinkIsSilent(t *testing.T) {
	repo := &captureRecordRepo{}
	ps := newRecordEntrypointTestService(requiredPicklistRecordEntrypointSchema(), repo)
	// No SetRejectedSaveAuditSink: recording is disabled.

	if _, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "probe"}, recordEntrypointUser()); err == nil {
		t.Fatal("Insert accepted a record missing a required picklist")
	}
	// Reaching here without a panic is the assertion.
}

func TestRejectedSaveAudit_KillSwitchStopsRecording(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink, rejectedSaveAuditDisabled: true}

	ps.emitRejectedSaveAudit(context.Background(), "Account", "001-rec", audit.ActionCreate,
		[]string{"Name"}, appErrors.NewValidationError("Name", "is required"), recordEntrypointUser())

	if sink.calls != 0 {
		t.Fatalf("kill switch is set but %d rows were emitted", sink.calls)
	}
}

func TestRejectedSaveAuditDisabledFromEnv(t *testing.T) {
	for _, tc := range []struct {
		raw          string
		wantDisabled bool
	}{
		{"", false},        // absent → recording ON, the #4261 default
		{"garbage", false}, // unparseable → fail safe to ON
		{"true", false},
		{"1", false},
		{"false", true}, // only an explicit false disables
		{"0", true},
	} {
		t.Run("value="+tc.raw, func(t *testing.T) {
			t.Setenv(rejectedSaveAuditEnabledEnv, tc.raw)
			if got := rejectedSaveAuditDisabledFromEnv(); got != tc.wantDisabled {
				t.Fatalf("rejectedSaveAuditDisabledFromEnv() = %v, want %v", got, tc.wantDisabled)
			}
		})
	}
}

// TestRejectedSaveAudit_SkipsSystemTables is the no-recursion assertion.
//
// Recursion is already impossible by construction — the sink writes raw SQL
// and never re-enters PersistenceService — but the system-table gate is the
// belt to that suspenders, and it keeps platform bookkeeping out of a log
// meant for user saves.
func TestRejectedSaveAudit_SkipsSystemTables(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink}

	ps.emitRejectedSaveAudit(context.Background(), constants.TableAuditLog, "row-1", audit.ActionCreate,
		[]string{"Outcome"}, appErrors.NewValidationError("Outcome", "is required"), recordEntrypointUser())

	if sink.calls != 0 {
		t.Fatalf("_System_AuditLog rejection emitted %d audit rows; the audit log must not audit itself", sink.calls)
	}
}

func TestRejectedSaveAudit_UnassignedRecordIdUsesSentinel(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink}

	// A create refused before Id assignment — for example FLS pruning, which
	// runs before generateSystemFields.
	ps.emitRejectedSaveAudit(context.Background(), "Account", "", audit.ActionCreate,
		[]string{"Secret__c"}, appErrors.NewFLSDeniedErrorForFields([]string{"Secret__c"}), recordEntrypointUser())

	event := sink.only(t)
	if event.RecordId != audit.RecordIdUnassigned {
		t.Fatalf("RecordId = %q, want %q — RecordId is NOT NULL and Validate() rejects empty",
			event.RecordId, audit.RecordIdUnassigned)
	}
}

// --- engine-internal writes are not user saves ------------------------------

// TestRejectedSaveAudit_SuppressesDerivedRecompute pins decision 2: a rollup
// recompute routed through Update is the platform reconciling a parent total,
// not a user pressing save. Recording it would put rows nobody can act on into
// a table support reads to answer "why was MY save refused".
func TestRejectedSaveAudit_SuppressesDerivedRecompute(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink}

	ctx := withDerivedRecompute(context.Background())
	ps.emitRejectedSaveAudit(ctx, "Account", "001-rec", audit.ActionUpdate,
		[]string{"Total__c"}, appErrors.NewValidationRuleError("Some_Rule", "blocked"),
		recordEntrypointUser())

	if sink.calls != 0 {
		t.Fatalf("derived recompute emitted %d rows; platform-internal writes are not user saves", sink.calls)
	}
}

// TestRejectedSaveAudit_SuppressesExpectedUniquenessProbeOnly pins decision 1,
// including its deliberate narrowness.
//
// The external-ID upsert attempts an INSERT and uses the uniqueness conflict
// as its signal that the record already exists, then adopts it via UPDATE — so
// that conflict describes a save that ultimately SUCCEEDS. Any OTHER refusal
// on the same attempt fails the whole upsert, the caller sees an error, and
// the row must still be written.
func TestRejectedSaveAudit_SuppressesExpectedUniquenessProbeOnly(t *testing.T) {
	for _, tc := range []struct {
		name     string
		err      error
		wantRows int
		wantWhy  string
	}{
		{
			name:     "uniqueness conflict is the probe working",
			err:      appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1"),
			wantRows: 0,
			wantWhy:  "the upsert adopts the existing record and succeeds",
		},
		{
			name:     "required field still fails the whole upsert",
			err:      appErrors.NewValidationError("Stage__c", "is required"),
			wantRows: 1,
			wantWhy:  "the caller sees an error, so the rejection is real",
		},
		{
			name:     "validation rule still fails the whole upsert",
			err:      appErrors.NewValidationRuleError("Block_Everything", "blocked"),
			wantRows: 1,
			wantWhy:  "the caller sees an error, so the rejection is real",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			sink := &captureFailureSink{}
			ps := &PersistenceService{rejectedSaveAuditSink: sink}

			ctx := withExpectedUniquenessConflict(context.Background())
			ps.emitRejectedSaveAudit(ctx, "Account", "001-rec", audit.ActionCreate,
				[]string{"ExternalId__c"}, tc.err, recordEntrypointUser())

			if sink.calls != tc.wantRows {
				t.Fatalf("rows = %d, want %d — %s", sink.calls, tc.wantRows, tc.wantWhy)
			}
		})
	}
}

// TestRejectedSaveAudit_UniquenessRecordedWithoutTheProbeMarker guards against
// the suppression being wider than intended: an ordinary uniqueness rejection,
// with no probe marker, is still recorded.
func TestRejectedSaveAudit_UniquenessRecordedWithoutTheProbeMarker(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink}

	ps.emitRejectedSaveAudit(context.Background(), "Account", "001-rec", audit.ActionCreate,
		[]string{"ExternalId__c"}, appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1"),
		recordEntrypointUser())

	if sink.calls != 1 {
		t.Fatalf("rows = %d, want 1: a plain uniqueness rejection is a real refused save", sink.calls)
	}
}

// TestRejectedSaveAudit_TerminalUpsertFailureIsStillRecorded closes the gap
// the probe suppression opened.
//
// Suppressing the attempted INSERT's uniqueness conflict is correct when the
// upsert goes on to adopt the existing record. But an adoption conflict or a
// lost unique race is TERMINAL — the caller gets an error and nothing landed.
// The suppression has already silenced the inner attempt and upsert has no
// other entry point, so without the outer hook those produce no row at all.
//
// The adoption-guard case is the one that nearly slipped: that error replaces
// the underlying conflict with a sentinel and keeps its diagnostics as text
// rather than as an error cause, so errors.As finds no *ConflictError and the
// generic branches do not classify it.
func TestRejectedSaveAudit_TerminalUpsertFailureIsStillRecorded(t *testing.T) {
	for _, tc := range []struct {
		name string
		err  error
	}{
		{
			name: "adoption guard conflict",
			err: newExternalIDUpsertAdoptionConflictError(
				"refusing to create or adopt Account because the requested guard identity is owned by another record",
				appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1")),
		},
		{
			name: "lost unique race",
			err:  fmt.Errorf("upsert lost the unique race: %w", appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1")),
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			// Sanity: this really is the shape that reaches the outer hook.
			if _, ok := classifyRejectedSave(tc.err); !ok {
				t.Fatalf("classifyRejectedSave(%T) = not-a-rejection; a terminal upsert failure would produce no row", tc.err)
			}

			sink := &captureFailureSink{}
			ps := &PersistenceService{rejectedSaveAuditSink: sink}
			ctx := withRejectedSaveClaim(context.Background())

			// The inner probe stays silent and must NOT take the claim, or the
			// outer hook would think the rejection was already recorded.
			ps.emitRejectedSaveAudit(withExpectedUniquenessConflict(ctx), "Account", "001-rec",
				audit.ActionCreate, []string{"ExternalId__c"},
				appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1"), recordEntrypointUser())
			if sink.calls != 0 {
				t.Fatalf("probe emitted %d rows; the expected uniqueness conflict must stay silent", sink.calls)
			}

			ps.emitRejectedSaveAudit(ctx, "Account", "", audit.ActionCreate,
				[]string{"ExternalId__c"}, tc.err, recordEntrypointUser())

			event := sink.only(t)
			if event.Outcome != audit.OutcomeFailure {
				t.Errorf("Outcome = %q, want FAILURE", event.Outcome)
			}
			changes := decodeChanges(t, event)
			if changes.RejectionStage != rejectionStageUniqueness {
				t.Fatalf("rejectionStage = %q, want %q", changes.RejectionStage, rejectionStageUniqueness)
			}
			// The guard's diagnostics are free text and may name the owning
			// record; only structural identity may be persisted.
			if strings.Contains(event.ChangesJSON, "ACME-1") {
				t.Fatalf("row leaked the conflicting value: %s", event.ChangesJSON)
			}
		})
	}
}

// TestRejectedSaveAudit_AdoptionWrapperDoesNotDoubleRecord covers the one case
// where identity-based claiming is not enough.
//
// The claim dedupes by error identity, which works because propagating frames
// share one error object. The adoption guard breaks that on purpose: its
// Unwrap returns a sentinel rather than the original cause, so the outer
// wrapper is NOT errors.Is the inner error. Without an explicit guard the
// inner hook records the conflict and the outer upsert hook records the
// wrapper too — two rows for one refused save.
func TestRejectedSaveAudit_AdoptionWrapperDoesNotDoubleRecord(t *testing.T) {
	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink}

	ctx := withRejectedSaveClaim(context.Background())
	claim := rejectedSaveClaimFrom(ctx)
	before := claim.recordedCount()

	inner := appErrors.NewConflictError("Account", "ExternalId__c", "ACME-1")
	wrapper := newExternalIDUpsertAdoptionConflictError("guard identity owned by another record", inner)

	// Sanity: the wrapper really does cut the cause chain, so identity
	// dedupe cannot see through it. If this stops being true the guard below
	// becomes redundant rather than wrong.
	if errors.Is(wrapper, inner) {
		t.Skip("adoption wrapper now preserves its cause; identity dedupe covers this")
	}

	// Inner frame (an Update inside the upsert) records the real conflict.
	ps.emitRejectedSaveAudit(ctx, "Account", "001-rec", audit.ActionUpdate,
		[]string{"ExternalId__c"}, inner, recordEntrypointUser())
	if sink.calls != 1 {
		t.Fatalf("inner hook emitted %d rows, want 1", sink.calls)
	}

	// Outer upsert hook: something was already recorded for this operation.
	if claim.recordedCount() > before {
		t.Log("outer hook correctly suppressed: an inner frame already recorded this operation")
	} else {
		ps.emitRejectedSaveAudit(ctx, "Account", "", audit.ActionCreate,
			[]string{"ExternalId__c"}, wrapper, recordEntrypointUser())
	}

	if sink.calls != 1 {
		t.Fatalf("emitted %d rows for one refused upsert, want 1: the adoption wrapper cuts the cause chain, "+
			"so identity dedupe alone lets the outer hook record it a second time", sink.calls)
	}
}

// TestRejectedSaveAudit_ProbeMarkerDoesNotLeakIntoNestedWrites pins the scope
// of the expected-uniqueness marker.
//
// The marker says "this ONE attempt's uniqueness conflict is planned". It
// lives on the context, and a context value reaches every nested write, so
// without an explicit clear a before-save flow that inserts a child record
// would have that child's REAL uniqueness conflict silenced too — reopening
// the blind spot underneath the suppression.
func TestRejectedSaveAudit_ProbeMarkerDoesNotLeakIntoNestedWrites(t *testing.T) {
	probeCtx := withExpectedUniquenessConflict(context.Background())
	if !expectsUniquenessConflict(probeCtx) {
		t.Fatal("marker did not take effect; the rest of this test would be vacuous")
	}

	nestedCtx := withoutExpectedUniquenessConflict(probeCtx)
	if expectsUniquenessConflict(nestedCtx) {
		t.Fatal("marker leaked into the derived context a nested write would receive")
	}

	sink := &captureFailureSink{}
	ps := &PersistenceService{rejectedSaveAuditSink: sink}
	conflict := appErrors.NewConflictError("Child", "Name", "dupe")

	// The owning attempt: still suppressed.
	ps.emitRejectedSaveAudit(probeCtx, "Account", "001-rec", audit.ActionCreate,
		[]string{"ExternalId__c"}, conflict, recordEntrypointUser())
	if sink.calls != 0 {
		t.Fatalf("the owning attempt emitted %d rows; its conflict is the planned one", sink.calls)
	}

	// A nested write's real conflict: recorded.
	ps.emitRejectedSaveAudit(nestedCtx, "Child", "002-rec", audit.ActionCreate,
		[]string{"Name"}, conflict, recordEntrypointUser())
	if sink.calls != 1 {
		t.Fatalf("nested write emitted %d rows, want 1: its uniqueness conflict is real and must be recorded", sink.calls)
	}
}