package records

// Nested-write attribution for the rejected-save audit (#4261).
//
// Regression pins for a defect these tests found by observation. Before the
// one-rejection-one-row claim in audit_rejected_save.go, a before-save handler
// that performed a NESTED write on a different object, refused, produced TWO
// rows: one correct row against the child, and one against the PARENT object
// carrying the CHILD's rejection identity. The mechanism is
// publishRecordEvent (persistence_events.go), which deliberately unwraps a
// typed business error out of the EventBus's InternalError envelope so REST
// can render it as a 400 — which also leaves it classifiable when it reaches
// the parent entry point's audit hook.
//
// The second test is the sharper one: a validation RULE name is not a field
// name, so the schema filter that keeps the required-field case honest does
// not apply. It observed `object=ProbeObject, rule=Child_Only_Rule` — a rule
// that does not exist on that object. Support querying the parent would chase
// a rule its metadata does not contain, which defeats the point of the
// feature more thoroughly than the original silence did.

import (
	"context"
	"strings"
	"testing"

	"github.com/nexuscrm/audit"
	eventFeature "github.com/nexuscrm/events"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedEvents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
)

// nestedWriteEventBus runs a caller-supplied nested write from inside the
// before-save publish, mimicking a record-triggered Flow whose RecordCreate
// element writes a different object.
type nestedWriteEventBus struct {
	on         sharedEvents.EventType
	onlyObject string
	nested     func(ctx context.Context) error
	fired      int
}

func (b *nestedWriteEventBus) Subscribe(_ sharedEvents.EventType, _ ports.EventHandler) func() {
	return func() {}
}

func (b *nestedWriteEventBus) Publish(ctx context.Context, eventType sharedEvents.EventType, payload interface{}) error {
	if eventType != b.on || b.nested == nil {
		return nil
	}
	recordPayload, ok := payload.(eventFeature.RecordEventPayload)
	if !ok {
		return nil
	}
	// Only the OUTER object's before-save fires the nested write; without
	// this the nested insert re-enters the same handler forever.
	if recordPayload.ObjectAPIName != b.onlyObject {
		return nil
	}
	b.fired++
	if err := b.nested(ctx); err != nil {
		// This is what the real EventBus does to a handler error, and what
		// publishRecordEvent then unwraps.
		return appErrors.NewInternalError("RecordBeforeCreate handler failed", err)
	}
	return nil
}

func (b *nestedWriteEventBus) PublishAsyncWithContext(context.Context, sharedEvents.EventType, interface{}) {
}

var _ ports.EventPublisher = (*nestedWriteEventBus)(nil)

// TestRejectedSaveAudit_NestedWriteAttribution observes how many rows a single
// user save produces when the rejection originates in a nested write, and what
// object each row names.
func TestRejectedSaveAudit_NestedWriteAttribution(t *testing.T) {
	// Child object: has a required picklist the nested write will omit.
	childSchema := requiredPicklistRecordEntrypointSchema()
	childSchema.APIName = "ChildProbe"
	makeFieldsMap(childSchema)

	// Parent object: nothing wrong with the parent's own payload.
	parentSchema := recordEntrypointSchema()

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

	// Point the metadata service at whichever schema is being asked for.
	ps.metadata = metadataFeature.NewMetadataService(
		&stubUpsertMetaRepo{schemas: []*models.ObjectMetadata{parentSchema, childSchema}}, nil, nil)
	ps.permissions = permissions.NewPermissionService(
		&stubUpsertPermRepo{grantCreate: true, grantEdit: true}, ps.metadata, nil)

	bus := &nestedWriteEventBus{on: sharedEvents.RecordBeforeCreate, onlyObject: "ProbeObject"}
	bus.nested = func(ctx context.Context) error {
		// The nested write: refused for a missing required picklist.
		_, err := ps.Insert(ctx, "ChildProbe",
			models.SObject{constants.FieldName: "child"}, recordEntrypointUser())
		return err
	}
	ps.eventBus = bus

	_, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "parent"}, recordEntrypointUser())

	if bus.fired == 0 {
		t.Fatal("before-save handler never ran; the fixture is not exercising a nested write")
	}
	if err == nil {
		t.Fatal("outer Insert succeeded despite the nested write being refused")
	}

	t.Logf("outer error: %v", err)
	t.Logf("rows emitted: %d", len(sink.events))
	for i, ev := range sink.events {
		t.Logf("  row %d: object=%s recordId=%s changes=%s", i, ev.ObjectApiName, ev.RecordId, ev.ChangesJSON)
	}

	// The observation that matters: is any row attributed to the PARENT while
	// naming the CHILD's rejection identity?
	var parentRows, childRows int
	for _, ev := range sink.events {
		switch ev.ObjectApiName {
		case "ProbeObject":
			parentRows++
		case "ChildProbe":
			childRows++
		}
	}
	if childRows != 1 {
		t.Errorf("child rows = %d, want 1: the nested rejection must be recorded against the object that refused it", childRows)
	}
	if parentRows > 0 {
		t.Errorf("parent rows = %d: a rejection that happened on ChildProbe was also recorded against ProbeObject, "+
			"which tells support a rule fired on an object that does not have it", parentRows)
	}
	if len(sink.events) != 1 {
		t.Errorf("total rows = %d, want 1 per rejected save", len(sink.events))
	}
	_ = audit.OutcomeFailure
}

// TestRejectedSaveAudit_NestedValidationRuleAttribution is the sharper variant:
// a validation-RULE name is not a field name, so the schema filter that keeps
// the required-field case honest does not apply to it. This observes whether
// the parent row ends up naming a rule that does not exist on the parent
// object.
func TestRejectedSaveAudit_NestedValidationRuleAttribution(t *testing.T) {
	childSchema := recordEntrypointSchema()
	childSchema.APIName = "ChildProbe"
	makeFieldsMap(childSchema)
	parentSchema := recordEntrypointSchema()

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

	// A rule that exists ONLY on ChildProbe.
	rules := []*models.ValidationRule{{
		ID:            "vr-child",
		ObjectAPIName: "ChildProbe",
		Name:          "Child_Only_Rule",
		Active:        true,
		Condition:     "true",
		ErrorMessage:  "child rule fired",
	}}
	ps.metadata = metadataFeature.NewMetadataService(
		&stubUpsertMetaRepo{
			schemas:         []*models.ObjectMetadata{parentSchema, childSchema},
			validationRules: rules,
		}, nil, nil)
	ps.permissions = permissions.NewPermissionService(
		&stubUpsertPermRepo{grantCreate: true, grantEdit: true}, ps.metadata, nil)

	bus := &nestedWriteEventBus{on: sharedEvents.RecordBeforeCreate, onlyObject: "ProbeObject"}
	bus.nested = func(ctx context.Context) error {
		_, err := ps.Insert(ctx, "ChildProbe",
			models.SObject{constants.FieldName: "child"}, recordEntrypointUser())
		return err
	}
	ps.eventBus = bus

	if _, err := ps.Insert(context.Background(), "ProbeObject",
		models.SObject{constants.FieldName: "parent"}, recordEntrypointUser()); err == nil {
		t.Fatal("outer Insert succeeded despite the nested write being refused")
	}
	if bus.fired == 0 {
		t.Fatal("before-save handler never ran")
	}

	for i, ev := range sink.events {
		t.Logf("  row %d: object=%s changes=%s", i, ev.ObjectApiName, ev.ChangesJSON)
	}
	for _, ev := range sink.events {
		if ev.ObjectApiName == "ProbeObject" && strings.Contains(ev.ChangesJSON, "Child_Only_Rule") {
			t.Errorf("parent row names a rule that exists only on ChildProbe: %s", ev.ChangesJSON)
		}
	}
}

// TestRejectedSaveAudit_CascadeRestoreAttributesToFailingRecord pins the
// cascade-restore half of the attribution rule.
//
// A restore is one user action but many record lifecycles: RestoreRecord's
// hook only knows the ROOT tombstone, while the transaction body runs every
// cascade tombstone's update. Unlike the nested-Insert case there is no child
// entry point to claim the rejection, so without the in-transaction claim the
// CHILD's failure lands on the PARENT record — support would look at the
// parent and find nothing wrong with it.
func TestRejectedSaveAudit_CascadeRestoreAttributesToFailingRecord(t *testing.T) {
	user := recordEntrypointUser()

	parentSchema := restoreObjectSchema("Parent__c")
	childSchema := restoreObjectSchema("Child__c")
	cascade := constants.DeleteRuleCascade
	childSchema.Fields = append(childSchema.Fields, models.FieldMetadata{
		APIName:     "Parent__c",
		Type:        constants.FieldTypeReference,
		ReferenceTo: []string{"Parent__c"},
		DeleteRule:  &cascade,
		Createable:  true,
		Updateable:  true,
	})
	// The child's Name is unique, and a LIVE row already holds the value the
	// restore would bring back — so the child, not the parent, is refused.
	for i := range childSchema.Fields {
		if childSchema.Fields[i].APIName == constants.FieldName {
			childSchema.Fields[i].IsUnique = true
		}
	}
	makeFieldsMap(childSchema)

	repo := newRecycleBinRestoreRepo("Parent__c", "parent-1", user.ID)
	operationID := "delete-op-audit"
	repo.tombstone = nil
	repo.tombstones = map[string]models.SObject{
		"parent-1": restoreTombstone("Parent__c", "parent-1", user.ID, models.SObject{
			constants.FieldSysRecycleBin_DeleteOperationID:       operationID,
			constants.FieldSysRecycleBin_DeleteRootObjectAPIName: "Parent__c",
			constants.FieldSysRecycleBin_DeleteRootRecordID:      "parent-1",
			constants.FieldSysRecycleBin_CascadeDepth:            0,
		}),
		"child-1": restoreTombstone("Child__c", "child-1", user.ID, models.SObject{
			constants.FieldSysRecycleBin_DeleteOperationID:          operationID,
			constants.FieldSysRecycleBin_DeleteRootObjectAPIName:    "Parent__c",
			constants.FieldSysRecycleBin_DeleteRootRecordID:         "parent-1",
			constants.FieldSysRecycleBin_CascadeParentObjectAPIName: "Parent__c",
			constants.FieldSysRecycleBin_CascadeParentRecordID:      "parent-1",
			constants.FieldSysRecycleBin_CascadeFieldAPIName:        "Parent__c",
			constants.FieldSysRecycleBin_CascadeDepth:               1,
		}),
	}
	repo.records["parent__c|parent-1"] = models.SObject{
		constants.FieldID:        "parent-1",
		constants.FieldName:      "Deleted parent",
		constants.FieldOwnerID:   user.ID,
		constants.FieldIsDeleted: true,
	}
	repo.records["child__c|child-1"] = models.SObject{
		constants.FieldID:        "child-1",
		constants.FieldName:      "Deleted child",
		constants.FieldOwnerID:   user.ID,
		constants.FieldIsDeleted: true,
		"Parent__c":              "parent-1",
	}
	repo.records["child__c|child-live"] = models.SObject{
		constants.FieldID:        "child-live",
		constants.FieldName:      "Deleted child",
		constants.FieldOwnerID:   user.ID,
		constants.FieldIsDeleted: false,
		"Parent__c":              "parent-1",
	}

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

	err := ps.RestoreRecord(context.Background(), "parent-1", user)
	if err == nil {
		t.Fatal("restore succeeded despite the child colliding on a unique field; fixture is not exercising a cascade rejection")
	}
	t.Logf("restore error: %v", err)
	for i, ev := range sink.events {
		t.Logf("  row %d: object=%s recordId=%s changes=%s", i, ev.ObjectApiName, ev.RecordId, ev.ChangesJSON)
	}
	if len(sink.events) != 1 {
		t.Fatalf("emitted %d rows, want 1 per rejected restore", len(sink.events))
	}

	ev := sink.events[0]
	if ev.ObjectApiName != "Child__c" || ev.RecordId != "child-1" {
		t.Fatalf("row names %s/%s, want Child__c/child-1 — the child is what was refused; "+
			"attributing it to the parent sends support to a record with nothing wrong with it",
			ev.ObjectApiName, ev.RecordId)
	}
	if ev.Action != audit.ActionUndelete {
		t.Errorf("Action = %q, want UNDELETE", ev.Action)
	}
}

// TestRejectedSaveAudit_ProbeMarkerClearedForNestedInsert is the end-to-end
// half of the marker-scope rule. The helper-level test proves
// withoutExpectedUniquenessConflict works; this proves Insert actually calls
// it, which is the part that would silently regress.
func TestRejectedSaveAudit_ProbeMarkerClearedForNestedInsert(t *testing.T) {
	parentSchema := recordEntrypointSchema()
	childSchema := recordEntrypointSchema()
	childSchema.APIName = "ChildProbe"
	for i := range childSchema.Fields {
		if childSchema.Fields[i].APIName == constants.FieldName {
			childSchema.Fields[i].IsUnique = true
		}
	}
	makeFieldsMap(childSchema)

	repo := &captureRecordRepo{
		stubUpsertRecordRepo: stubUpsertRecordRepo{
			checkUniqueFn: func(_ context.Context, schema *models.ObjectMetadata, _ string, _ interface{}, _ string) (bool, error) {
				// Only the CHILD collides — a real conflict that must be recorded.
				return schema != nil && schema.APIName == "ChildProbe", nil
			},
		},
	}
	ps, sink := newAuditedEntrypointService(parentSchema, repo)
	ps.metadata = metadataFeature.NewMetadataService(
		&stubUpsertMetaRepo{schemas: []*models.ObjectMetadata{parentSchema, childSchema}}, nil, nil)
	ps.permissions = permissions.NewPermissionService(
		&stubUpsertPermRepo{grantCreate: true, grantEdit: true}, ps.metadata, nil)

	bus := &nestedWriteEventBus{on: sharedEvents.RecordBeforeCreate, onlyObject: "ProbeObject"}
	bus.nested = func(ctx context.Context) error {
		_, err := ps.Insert(ctx, "ChildProbe",
			models.SObject{constants.FieldName: "collides"}, recordEntrypointUser())
		return err
	}
	ps.eventBus = bus

	// The OUTER insert carries the probe marker, as the external-ID upsert's
	// attempted insert does.
	ctx := withExpectedUniquenessConflict(context.Background())
	if _, err := ps.Insert(ctx, "ProbeObject",
		models.SObject{constants.FieldName: "parent"}, recordEntrypointUser()); err == nil {
		t.Fatal("outer Insert succeeded despite the nested child colliding")
	}
	if bus.fired == 0 {
		t.Fatal("before-save handler never ran; the fixture is not exercising a nested write")
	}

	for i, ev := range sink.events {
		t.Logf("  row %d: object=%s changes=%s", i, ev.ObjectApiName, ev.ChangesJSON)
	}
	if len(sink.events) != 1 || sink.events[0].ObjectApiName != "ChildProbe" {
		t.Fatalf("got %d rows %v; want exactly one on ChildProbe — the probe marker belongs to the outer "+
			"insert only, and the nested write's uniqueness conflict is real",
			len(sink.events), sink.events)
	}
}