package automation

import (
	"context"
	"fmt"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/api"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

type fanoutTxKey struct{}
type fanoutPessimisticKey struct{}

type fanoutInsert struct {
	object string
	record models.SObject
}

type transactionalFanoutPersistence struct {
	failOnObject      string
	nextID            int
	runTxCalls        int
	runPessTxCalls    int
	sawAtomicBoundary bool
	insertPrelocks    [][]recordlocks.ParentKey
	committed         []fanoutInsert
}

type fanoutSourceChildQuery struct {
	records []models.SObject
	calls   int
}

func (q *fanoutSourceChildQuery) QueryWithFilter(context.Context, string, string, *models.UserSession, string, string, int) (*models.QueryResult, error) {
	q.calls++
	out := make([]models.SObject, 0, len(q.records))
	for _, record := range q.records {
		out = append(out, cloneFanoutSObject(record))
	}
	return &models.QueryResult{
		TotalSize: len(out),
		Done:      true,
		Records:   out,
	}, nil
}

func (q *fanoutSourceChildQuery) Query(context.Context, api.QueryRequest, *models.UserSession) (*models.QueryResult, error) {
	return nil, fmt.Errorf("unexpected Query call in fan-out test")
}

func (q *fanoutSourceChildQuery) QueryWithSpec(context.Context, queryspec.QuerySpec, *models.UserSession) (*models.QueryResult, error) {
	return nil, fmt.Errorf("unexpected QueryWithSpec call in fan-out test")
}

func (q *fanoutSourceChildQuery) RunAnalytics(context.Context, models.AnalyticsQuery, *models.UserSession) (interface{}, error) {
	return nil, fmt.Errorf("unexpected RunAnalytics call in fan-out test")
}

func (q *fanoutSourceChildQuery) SearchSingleObject(context.Context, string, string, []string, []string, int, int, string, string, string, *models.UserSession) ([]models.SObject, error) {
	return nil, fmt.Errorf("unexpected SearchSingleObject call in fan-out test")
}

func (q *fanoutSourceChildQuery) SearchSuggestions(context.Context, string, string, []string, []string, int, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (q *fanoutSourceChildQuery) GlobalSearch(context.Context, string, *models.UserSession) ([]models.SearchResult, error) {
	return nil, fmt.Errorf("unexpected GlobalSearch call in fan-out test")
}

func (q *fanoutSourceChildQuery) Calculate(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	return cloneFanoutSObject(record), nil
}

func (q *fanoutSourceChildQuery) ExecuteRawSQL(context.Context, string, []interface{}, *models.UserSession) ([]models.SObject, error) {
	return nil, fmt.Errorf("unexpected ExecuteRawSQL call in fan-out test")
}

type fanoutTx struct {
	inserts []fanoutInsert
}

func (p *transactionalFanoutPersistence) Insert(ctx context.Context, objectName string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	p.insertPrelocks = append(p.insertPrelocks, recordlocks.ForcedParentPrelocksFromContext(ctx))
	if p.failOnObject != "" && objectName == p.failOnObject {
		return nil, fmt.Errorf("synthetic create failure for %s", objectName)
	}
	if isWithinAtomicFlowBoundary(ctx) {
		p.sawAtomicBoundary = true
	}

	p.nextID++
	out := cloneFanoutSObject(record)
	out[constants.FieldID] = fmt.Sprintf("%s-%d", objectName, p.nextID)
	insert := fanoutInsert{object: objectName, record: cloneFanoutSObject(out)}
	if tx, ok := ctx.Value(fanoutTxKey{}).(*fanoutTx); ok {
		tx.inserts = append(tx.inserts, insert)
		return out, nil
	}
	p.committed = append(p.committed, insert)
	return out, nil
}

func (p *transactionalFanoutPersistence) PreviewBeforeInsert(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	return cloneFanoutSObject(record), nil
}

func (p *transactionalFanoutPersistence) PrelockUpdateParents(context.Context, string, string, models.SObject, *models.UserSession) error {
	return nil
}

func (p *transactionalFanoutPersistence) Update(context.Context, string, string, models.SObject, *models.UserSession) error {
	return nil
}

func (p *transactionalFanoutPersistence) PreviewBeforeUpdate(context.Context, string, string, models.SObject, models.SObject, *models.UserSession) (*models.PreparedRecordUpdate, error) {
	return &models.PreparedRecordUpdate{}, nil
}

func (p *transactionalFanoutPersistence) CommitPreparedUpdate(context.Context, string, string, *models.PreparedRecordUpdate, models.SObject, *models.UserSession) error {
	return nil
}

func (p *transactionalFanoutPersistence) Delete(context.Context, string, string, *models.UserSession) error {
	return nil
}

func (p *transactionalFanoutPersistence) GetRaw(context.Context, string, string) (models.SObject, error) {
	return nil, nil
}

func (p *transactionalFanoutPersistence) GetLock(context.Context, string, string) (models.SObject, error) {
	return nil, nil
}

func (p *transactionalFanoutPersistence) GetRecordType(context.Context, string) (string, error) {
	return "", nil
}

func (p *transactionalFanoutPersistence) RunInTransaction(ctx context.Context, fn func(context.Context) error) error {
	p.runTxCalls++
	tx := &fanoutTx{}
	txCtx := context.WithValue(ctx, fanoutTxKey{}, tx)
	txCtx = context.WithValue(txCtx, fanoutPessimisticKey{}, false)
	err := fn(txCtx)
	if err != nil {
		return err
	}
	p.committed = append(p.committed, tx.inserts...)
	return nil
}

func (p *transactionalFanoutPersistence) IsTransactionActive(ctx context.Context) bool {
	tx, ok := ctx.Value(fanoutTxKey{}).(*fanoutTx)
	return ok && tx != nil
}

func (p *transactionalFanoutPersistence) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	if tx, ok := ctx.Value(fanoutTxKey{}).(*fanoutTx); ok && tx != nil {
		if pessimistic, _ := ctx.Value(fanoutPessimisticKey{}).(bool); pessimistic {
			p.runPessTxCalls++
			return fn(ctx)
		}
		return fmt.Errorf("pessimistic transaction requires owned outer boundary")
	}

	p.runPessTxCalls++
	tx := &fanoutTx{}
	txCtx := context.WithValue(ctx, fanoutTxKey{}, tx)
	txCtx = context.WithValue(txCtx, fanoutPessimisticKey{}, true)
	err := fn(txCtx)
	if err != nil {
		return err
	}
	p.committed = append(p.committed, tx.inserts...)
	return nil
}

func (p *transactionalFanoutPersistence) DetachTx(ctx context.Context) context.Context {
	return ctx
}

func (p *transactionalFanoutPersistence) RecomputeActiveFormulaProjection(_ context.Context, _ *models.UserSession, _, _ string, _ []string) (models.SObject, error) {
	return nil, nil
}

func cloneFanoutSObject(in models.SObject) models.SObject {
	out := models.SObject{}
	for k, v := range in {
		out[k] = v
	}
	return out
}

func fanoutExecutor(meta models.FlowMetadata, persistence *transactionalFanoutPersistence) *FlowExecutor {
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:          persistence,
		logger:               logging.New(context.Background()),
		formulaEngine:        eng,
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(eng),
	}
	exec.interviews = NewFlowInterviewService(persistence, nil)
	exec.flowVersionFn = func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:           "synthetic-fanout-version",
			DefinitionId: "synthetic-fanout-definition",
			Source:       "custom",
			Status:       "Active",
			Metadata:     meta,
		}, nil
	}
	return exec
}

func syntheticFanoutFlow(runAtomically bool) models.FlowMetadata {
	return models.FlowMetadata{
		RunAtomically: runAtomically,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "SyntheticParent",
				AssignRecordIdToReference: "newParentId",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("copied parent")},
				},
				Connector: &models.FlowConnector{TargetReference: "create_first_child"},
			},
			{
				Name:   "create_first_child",
				Object: "SyntheticChild",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("first child")},
					{Field: "ParentId", Value: elementReferenceValue("newParentId")},
				},
				Connector: &models.FlowConnector{TargetReference: "create_failing_child"},
			},
			{
				Name:   "create_failing_child",
				Object: "SyntheticFailingChild",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("second child")},
					{Field: "ParentId", Value: elementReferenceValue("newParentId")},
				},
			},
		},
	}
}

func syntheticLookupLoopFanoutFlow() models.FlowMetadata {
	return models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "SyntheticParent",
				AssignRecordIdToReference: "newParentId",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("copied parent")},
				},
				Connector: &models.FlowConnector{TargetReference: "lookup_source_children"},
			},
			{
				Name:   "create_child",
				Object: "SyntheticChild",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: elementReferenceValue("currentSourceChild.Name")},
					{Field: "SourceChildId", Value: elementReferenceValue("currentSourceChild.Id")},
					{Field: "ParentId", Value: elementReferenceValue("newParentId")},
				},
				Connector: &models.FlowConnector{TargetReference: "loop_source_children"},
			},
		},
		RecordLookups: []models.FlowRecordLookup{{
			Name:                     "lookup_source_children",
			Object:                   "SyntheticSourceChild",
			OutputReference:          "sourceChildren",
			StoreOutputAutomatically: true,
			GetFirstRecordOnly:       false,
			Connector:                &models.FlowConnector{TargetReference: "loop_source_children"},
		}},
		Loops: []models.FlowLoop{{
			Name:                       "loop_source_children",
			CollectionReference:        "sourceChildren",
			AssignNextValueToReference: "currentSourceChild",
			NextValueConnector:         &models.FlowConnector{TargetReference: "create_child"},
		}},
	}
}

func syntheticNestedCollectionLoopFanoutFlow() models.FlowMetadata {
	return models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		Variables: []models.FlowVariable{
			{Name: "payload", DataType: "SObject", IsInput: true},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "SyntheticParent",
				AssignRecordIdToReference: "newParentId",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("copied parent")},
				},
				Connector: &models.FlowConnector{TargetReference: "loop_payload_children"},
			},
			{
				Name:   "create_child",
				Object: "SyntheticChild",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: elementReferenceValue("currentPayloadChild.Name")},
					{Field: "SourceChildId", Value: elementReferenceValue("currentPayloadChild.Id")},
					{Field: "ParentId", Value: elementReferenceValue("newParentId")},
				},
				Connector: &models.FlowConnector{TargetReference: "loop_payload_children"},
			},
		},
		Loops: []models.FlowLoop{{
			Name:                       "loop_payload_children",
			CollectionReference:        "payload.sourceChildren",
			AssignNextValueToReference: "currentPayloadChild",
			NextValueConnector:         &models.FlowConnector{TargetReference: "create_child"},
		}},
	}
}

func syntheticNestedScalarLoopReferenceFlow() models.FlowMetadata {
	return models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		Variables: []models.FlowVariable{
			{Name: "payload", DataType: "SObject", IsInput: true},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "SyntheticParent",
				AssignRecordIdToReference: "newParentId",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("copied parent")},
				},
				Connector: &models.FlowConnector{TargetReference: "loop_payload_children"},
			},
			{
				Name:   "create_child",
				Object: "SyntheticChild",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: elementReferenceValue("currentPayloadChild.Name")},
					{Field: "ParentId", Value: elementReferenceValue("newParentId")},
				},
				Connector: &models.FlowConnector{TargetReference: "loop_payload_children"},
			},
		},
		Loops: []models.FlowLoop{{
			Name:                       "loop_payload_children",
			CollectionReference:        "payload.sourceChild.Name",
			AssignNextValueToReference: "currentPayloadChild",
			NextValueConnector:         &models.FlowConnector{TargetReference: "create_child"},
		}},
	}
}

func syntheticNestedCollectionLoopMutationFlow() models.FlowMetadata {
	return models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "create_parent"},
		},
		Variables: []models.FlowVariable{
			{Name: "payload", DataType: "SObject", IsInput: true},
			{Name: "replacementChildren", DataType: "SObject", IsCollection: true, IsInput: true},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "create_parent",
				Object:                    "SyntheticParent",
				AssignRecordIdToReference: "newParentId",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: stringValue("copied parent")},
				},
				Connector: &models.FlowConnector{TargetReference: "loop_payload_children"},
			},
			{
				Name:   "create_child",
				Object: "SyntheticChild",
				InputAssignments: []models.FlowInputAssignment{
					{Field: "Name", Value: elementReferenceValue("currentPayloadChild.Name")},
					{Field: "SourceChildId", Value: elementReferenceValue("currentPayloadChild.Id")},
					{Field: "ParentId", Value: elementReferenceValue("newParentId")},
				},
				Connector: &models.FlowConnector{TargetReference: "replace_payload_children"},
			},
		},
		Assignments: []models.FlowAssignment{
			{
				Name:      "replace_payload_children",
				Connector: &models.FlowConnector{TargetReference: "loop_payload_children"},
				Items: []models.FlowAssignmentItem{
					{
						AssignToReference: "payload.sourceChildren",
						Operator:          "Assign",
						Value:             elementReferenceValue("replacementChildren"),
					},
				},
			},
		},
		Loops: []models.FlowLoop{{
			Name:                       "loop_payload_children",
			CollectionReference:        "payload.sourceChildren",
			AssignNextValueToReference: "currentPayloadChild",
			NextValueConnector:         &models.FlowConnector{TargetReference: "create_child"},
		}},
	}
}

func TestLookupLoopRecordCreateFanoutHandlesZeroOneManySourceChildren(t *testing.T) {
	for _, tc := range []struct {
		name        string
		childNames  []string
		wantCreates int
	}{
		{name: "zero", childNames: nil, wantCreates: 1},
		{name: "one", childNames: []string{"first child"}, wantCreates: 2},
		{name: "many", childNames: []string{"first child", "second child", "third child"}, wantCreates: 4},
	} {
		t.Run(tc.name, func(t *testing.T) {
			persistence := &transactionalFanoutPersistence{}
			query := &fanoutSourceChildQuery{records: sourceChildRows(tc.childNames)}
			exec := fanoutExecutor(syntheticLookupLoopFanoutFlow(), persistence)
			exec.query = query
			exec.interviews = NewFlowInterviewService(persistence, query)

			result, err := exec.ExecuteFlow(WithTransientFlowInterviews(context.Background()), "SyntheticCreateFromSource", nil, newPreviewUser())
			if err != nil {
				t.Fatalf("ExecuteFlow returned error: %v", err)
			}
			if result == nil || !result.IsSuccess {
				t.Fatalf("ExecuteFlow result = %#v; want success", result)
			}
			if query.calls != 1 {
				t.Fatalf("QueryWithFilter calls = %d; want 1 source-child lookup", query.calls)
			}
			if got := len(persistence.committed); got != tc.wantCreates {
				t.Fatalf("committed inserts = %d (%+v); want %d", got, persistence.committed, tc.wantCreates)
			}
			parent := persistence.committed[0]
			if parent.object != "SyntheticParent" {
				t.Fatalf("first committed object = %q; want SyntheticParent", parent.object)
			}
			parentID := parent.record[constants.FieldID]
			for idx, wantName := range tc.childNames {
				child := persistence.committed[idx+1]
				if child.object != "SyntheticChild" {
					t.Fatalf("committed[%d].object = %q; want SyntheticChild", idx+1, child.object)
				}
				if got := child.record["Name"]; got != wantName {
					t.Fatalf("child %d Name = %v; want %q", idx, got, wantName)
				}
				if got := child.record["SourceChildId"]; got != fmt.Sprintf("source-child-%d", idx+1) {
					t.Fatalf("child %d SourceChildId = %v", idx, got)
				}
				if got := child.record["ParentId"]; got != parentID {
					t.Fatalf("child %d ParentId = %v; want parent id %v", idx, got, parentID)
				}
			}
		})
	}
}

func TestLoopRecordCreateFanoutHandlesNestedCollectionReference(t *testing.T) {
	persistence := &transactionalFanoutPersistence{}
	exec := fanoutExecutor(syntheticNestedCollectionLoopFanoutFlow(), persistence)

	payloadChildren := []interface{}{
		map[string]interface{}{"Id": "payload-child-1", "Name": "first child"},
		map[string]interface{}{"Id": "payload-child-2", "Name": "second child"},
	}

	result, err := exec.ExecuteFlow(
		WithTransientFlowInterviews(context.Background()),
		"SyntheticCreateFromNestedPayload",
		map[string]interface{}{
			"payload": map[string]interface{}{
				"sourceChildren": payloadChildren,
			},
		},
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlow returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("ExecuteFlow result = %#v; want success", result)
	}
	if got, want := len(persistence.committed), 3; got != want {
		t.Fatalf("committed inserts = %d (%+v); want %d", got, persistence.committed, want)
	}
	parent := persistence.committed[0]
	parentID := parent.record[constants.FieldID]
	for idx, wantName := range []string{"first child", "second child"} {
		child := persistence.committed[idx+1]
		if child.object != "SyntheticChild" {
			t.Fatalf("committed[%d].object = %q; want SyntheticChild", idx+1, child.object)
		}
		if got := child.record["Name"]; got != wantName {
			t.Fatalf("child %d Name = %v; want %q", idx, got, wantName)
		}
		if got := child.record["SourceChildId"]; got != fmt.Sprintf("payload-child-%d", idx+1) {
			t.Fatalf("child %d SourceChildId = %v", idx, got)
		}
		if got := child.record["ParentId"]; got != parentID {
			t.Fatalf("child %d ParentId = %v; want parent id %v", idx, got, parentID)
		}
	}
}

func TestLoopRecordCreateFanoutIgnoresCallerSuppliedLoopRuntimeKeys(t *testing.T) {
	persistence := &transactionalFanoutPersistence{}
	exec := fanoutExecutor(syntheticNestedCollectionLoopFanoutFlow(), persistence)

	payloadChildren := []interface{}{
		map[string]interface{}{"Id": "payload-child-1", "Name": "first child"},
		map[string]interface{}{"Id": "payload-child-2", "Name": "second child"},
	}

	result, err := exec.ExecuteFlow(
		WithTransientFlowInterviews(context.Background()),
		"SyntheticCreateFromNestedPayload",
		map[string]interface{}{
			"payload": map[string]interface{}{
				"sourceChildren": payloadChildren,
			},
			"_loop_loop_payload_children_items": []interface{}{
				map[string]interface{}{"Id": "evil-child-1", "Name": "evil child"},
			},
			"_loop_loop_payload_children_index": 1,
		},
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlow returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("ExecuteFlow result = %#v; want success", result)
	}
	if got, want := len(persistence.committed), 3; got != want {
		t.Fatalf("committed inserts = %d (%+v); want %d", got, persistence.committed, want)
	}
	parentID := persistence.committed[0].record[constants.FieldID]
	for idx, wantName := range []string{"first child", "second child"} {
		child := persistence.committed[idx+1]
		if got := child.record["Name"]; got != wantName {
			t.Fatalf("child %d Name = %v; want %q", idx, got, wantName)
		}
		if got := child.record["SourceChildId"]; got != fmt.Sprintf("payload-child-%d", idx+1) {
			t.Fatalf("child %d SourceChildId = %v", idx, got)
		}
		if got := child.record["ParentId"]; got != parentID {
			t.Fatalf("child %d ParentId = %v; want parent id %v", idx, got, parentID)
		}
	}
}

func TestLoopRecordCreateFanoutRereadsMutatedNestedCollectionReference(t *testing.T) {
	persistence := &transactionalFanoutPersistence{}
	exec := fanoutExecutor(syntheticNestedCollectionLoopMutationFlow(), persistence)

	payloadChildren := []interface{}{
		map[string]interface{}{"Id": "payload-child-1", "Name": "first child"},
		map[string]interface{}{"Id": "payload-child-2", "Name": "second child"},
		map[string]interface{}{"Id": "payload-child-3", "Name": "third child"},
	}
	replacementChildren := []interface{}{
		map[string]interface{}{"Id": "payload-child-1", "Name": "first child"},
		map[string]interface{}{"Id": "payload-child-replacement", "Name": "replacement child"},
	}

	result, err := exec.ExecuteFlow(
		WithTransientFlowInterviews(context.Background()),
		"SyntheticCreateFromNestedPayloadMutation",
		map[string]interface{}{
			"payload": map[string]interface{}{
				"sourceChildren": payloadChildren,
			},
			"replacementChildren": replacementChildren,
		},
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlow returned error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("ExecuteFlow result = %#v; want success", result)
	}
	if got, want := len(persistence.committed), 3; got != want {
		t.Fatalf("committed inserts = %d (%+v); want %d so loop follows the replacement collection on the next visit", got, persistence.committed, want)
	}
	parentID := persistence.committed[0].record[constants.FieldID]
	wantChildren := []struct {
		name string
		id   string
	}{
		{name: "first child", id: "payload-child-1"},
		{name: "replacement child", id: "payload-child-replacement"},
	}
	for idx, want := range wantChildren {
		child := persistence.committed[idx+1]
		if got := child.record["Name"]; got != want.name {
			t.Fatalf("child %d Name = %v; want %q", idx, got, want.name)
		}
		if got := child.record["SourceChildId"]; got != want.id {
			t.Fatalf("child %d SourceChildId = %v; want %q", idx, got, want.id)
		}
		if got := child.record["ParentId"]; got != parentID {
			t.Fatalf("child %d ParentId = %v; want parent id %v", idx, got, parentID)
		}
	}
}

func TestLoopRecordCreateFanoutFailsClosedForNestedScalarCollectionReference(t *testing.T) {
	persistence := &transactionalFanoutPersistence{}
	exec := fanoutExecutor(syntheticNestedScalarLoopReferenceFlow(), persistence)

	result, err := exec.ExecuteFlow(
		WithTransientFlowInterviews(context.Background()),
		"SyntheticCreateFromNestedScalarPayload",
		map[string]interface{}{
			"payload": map[string]interface{}{
				"sourceChild": map[string]interface{}{
					"Id":   "payload-child-1",
					"Name": "not-a-collection",
				},
			},
		},
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlow returned error: %v", err)
	}
	if result == nil {
		t.Fatal("ExecuteFlow returned nil result")
	}
	if result.IsSuccess {
		t.Fatalf("ExecuteFlow result = %#v; want invalid-flow failure for scalar loop collection reference", result)
	}
	if got := len(persistence.committed); got != 0 {
		t.Fatalf("committed inserts = %d (%+v); want 0 because scalar loop collection references must fail closed before commit", got, persistence.committed)
	}
	if len(result.Errors) == 0 {
		t.Fatalf("flow errors = %+v; want INVALID_FLOW payload", result.Errors)
	}
	if got := result.Errors[0].StatusCode; got != statusCodeInvalidFlow {
		t.Fatalf("first error statusCode = %q; want %q", got, statusCodeInvalidFlow)
	}
	if !strings.Contains(result.Errors[0].Message, "payload.sourceChild.Name") {
		t.Fatalf("first error message = %q; want collection reference context", result.Errors[0].Message)
	}
}

func sourceChildRows(names []string) []models.SObject {
	rows := make([]models.SObject, 0, len(names))
	for idx, name := range names {
		rows = append(rows, models.SObject{
			constants.FieldID: fmt.Sprintf("source-child-%d", idx+1),
			"Name":            name,
		})
	}
	return rows
}

func TestRunAtomicallyRollsBackParentAndChildCreatesOnFanoutFailure(t *testing.T) {
	persistence := &transactionalFanoutPersistence{failOnObject: "SyntheticFailingChild"}
	exec := fanoutExecutor(syntheticFanoutFlow(true), persistence)

	result, err := exec.ExecuteFlow(WithTransientFlowInterviews(context.Background()), "SyntheticCreateFromSource", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("ExecuteFlow returned error: %v", err)
	}
	if result == nil {
		t.Fatal("ExecuteFlow returned nil result")
	}
	if result.IsSuccess {
		t.Fatalf("IsSuccess=true; want failed flow when child fan-out create fails")
	}
	if got := len(persistence.committed); got != 0 {
		t.Fatalf("committed inserts = %d (%+v); want 0 because parent and first child must roll back together", got, persistence.committed)
	}
	if persistence.runPessTxCalls != 1 {
		t.Fatalf("RunInPessimisticTransaction calls = %d; want 1", persistence.runPessTxCalls)
	}
	if persistence.runTxCalls != 0 {
		t.Fatalf("RunInTransaction calls = %d; want 0", persistence.runTxCalls)
	}
	if !persistence.sawAtomicBoundary {
		t.Fatal("recordCreate did not observe atomic Flow boundary")
	}
	if len(result.Errors) == 0 || !strings.Contains(result.Errors[0].Message, "SyntheticFailingChild") {
		t.Fatalf("flow errors = %+v; want child create failure surfaced", result.Errors)
	}
}

func TestNonAtomicFanoutFailureDoesNotPretendToRollbackCommittedCreates(t *testing.T) {
	persistence := &transactionalFanoutPersistence{failOnObject: "SyntheticFailingChild"}
	exec := fanoutExecutor(syntheticFanoutFlow(false), persistence)

	result, err := exec.ExecuteFlow(WithTransientFlowInterviews(context.Background()), "SyntheticCreateFromSource", nil, newPreviewUser())
	if err != nil {
		t.Fatalf("ExecuteFlow returned error: %v", err)
	}
	if result == nil || result.IsSuccess {
		t.Fatalf("ExecuteFlow result = %#v; want failed flow", result)
	}
	if persistence.runTxCalls != 0 {
		t.Fatalf("RunInTransaction calls = %d; want 0 for non-atomic flow", persistence.runTxCalls)
	}
	if persistence.runPessTxCalls != 0 {
		t.Fatalf("RunInPessimisticTransaction calls = %d; want 0 for non-atomic flow", persistence.runPessTxCalls)
	}
	if got, want := len(persistence.committed), 2; got != want {
		t.Fatalf("committed inserts = %d (%+v); want %d for non-atomic contrast", got, persistence.committed, want)
	}
	if persistence.committed[0].object != "SyntheticParent" {
		t.Fatalf("first committed object = %q; want SyntheticParent", persistence.committed[0].object)
	}
	if parentID := persistence.committed[0].record[constants.FieldID]; persistence.committed[1].record["ParentId"] != parentID {
		t.Fatalf("child ParentId = %v; want parent id %v", persistence.committed[1].record["ParentId"], parentID)
	}
}

func TestRunAtomicallyFailsClosedInsideForeignTransactionWithoutCommittingWrites(t *testing.T) {
	persistence := &transactionalFanoutPersistence{failOnObject: "SyntheticFailingChild"}
	exec := fanoutExecutor(syntheticFanoutFlow(true), persistence)

	err := persistence.RunInTransaction(WithTransientFlowInterviews(context.Background()), func(txCtx context.Context) error {
		result, execErr := exec.ExecuteFlow(txCtx, "SyntheticCreateFromSource", nil, newPreviewUser())
		if execErr != nil {
			return execErr
		}
		if result == nil || result.IsSuccess {
			t.Fatalf("ExecuteFlow result = %#v; want failed flow when child create fails", result)
		}
		if len(result.Errors) == 0 || !strings.Contains(result.Errors[0].Message, "owned rollback boundary") {
			t.Fatalf("flow errors = %+v; want fail-closed foreign-tx error", result.Errors)
		}
		return nil
	})
	if err != nil {
		t.Fatalf("outer RunInTransaction: %v", err)
	}
	if got := len(persistence.committed); got != 0 {
		t.Fatalf("committed inserts = %d (%+v); want 0 because fail-closed path must not land partial writes", got, persistence.committed)
	}
	if persistence.runTxCalls != 1 {
		t.Fatalf("RunInTransaction calls = %d; want 1 outer call", persistence.runTxCalls)
	}
	if persistence.runPessTxCalls != 0 {
		t.Fatalf("RunInPessimisticTransaction calls = %d; want 0 on fail-closed foreign-tx path", persistence.runPessTxCalls)
	}
	if persistence.sawAtomicBoundary {
		t.Fatal("recordCreate observed atomic boundary; want flow body to never start on fail-closed path")
	}
}