package automation

import (
	"context"
	"errors"
	"testing"

	"github.com/nexuscrm/backend/internal/domain/ports"
	"github.com/nexuscrm/backend/pkg/logging"
	"github.com/nexuscrm/shared/pkg/api"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
)

// ---------------------------------------------------------------------------
// MockQueryService for executor tests
// ---------------------------------------------------------------------------

type MockQueryService struct {
	mock.Mock
}

func (m *MockQueryService) Query(ctx context.Context, req api.QueryRequest, user *models.UserSession) (*models.QueryResult, error) {
	args := m.Called(ctx, req, user)
	if args.Get(0) == nil {
		return nil, args.Error(1)
	}
	return args.Get(0).(*models.QueryResult), args.Error(1)
}

func (m *MockQueryService) QueryWithFilter(ctx context.Context, objectName string, filterExpr string, user *models.UserSession, sortField string, sortDir string, limit int) (*models.QueryResult, error) {
	args := m.Called(ctx, objectName, filterExpr, user, sortField, sortDir, limit)
	if args.Get(0) == nil {
		return nil, args.Error(1)
	}
	return args.Get(0).(*models.QueryResult), args.Error(1)
}

func (m *MockQueryService) RunAnalytics(ctx context.Context, query models.AnalyticsQuery, user *models.UserSession) (interface{}, error) {
	return nil, nil
}

func (m *MockQueryService) SearchSingleObject(ctx context.Context, objectName string, term string, searchFields []string, returnFields []string, limit int, offset int, orderBy string, orderDir string, filterExpr string, user *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (m *MockQueryService) GlobalSearch(ctx context.Context, term string, user *models.UserSession) ([]models.SearchResult, error) {
	return nil, nil
}

func (m *MockQueryService) Calculate(ctx context.Context, objectName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
	return nil, nil
}

func (m *MockQueryService) ExecuteRawSQL(ctx context.Context, sql string, params []interface{}, user *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func TestActiveFlowIsPackOwnedRequiresPackDefinitionAndStandardSource(t *testing.T) {
	t.Run("customized definition is not pack-owned", func(t *testing.T) {
		query := new(MockQueryService)
		executor := &FlowExecutor{query: query}
		query.On("Query",
			mock.Anything,
			mock.MatchedBy(func(req api.QueryRequest) bool {
				return req.ObjectAPIName == constants.TableFlowDefinition
			}),
			mock.Anything,
		).Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{{
				"DeveloperName":   "ChildFlow",
				"ActiveVersionId": "flow-child-v1",
				"IsCustomized":    true,
			}},
		}, nil).Once()

		got, err := executor.activeFlowIsPackOwned(context.Background(), "ChildFlow", nil)
		assert.NoError(t, err)
		assert.False(t, got)
		query.AssertExpectations(t)
	})

	t.Run("custom active flow source is not pack-owned", func(t *testing.T) {
		query := new(MockQueryService)
		executor := &FlowExecutor{query: query}
		query.On("Query",
			mock.Anything,
			mock.MatchedBy(func(req api.QueryRequest) bool {
				return req.ObjectAPIName == constants.TableFlowDefinition
			}),
			mock.Anything,
		).Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{{
				"DeveloperName":   "ChildFlow",
				"ActiveVersionId": "flow-child-v1",
				"IsCustomized":    false,
			}},
		}, nil).Once()
		query.On("Query",
			mock.Anything,
			mock.MatchedBy(func(req api.QueryRequest) bool {
				return req.ObjectAPIName == constants.TableFlow
			}),
			mock.Anything,
		).Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{{
				"Id":     "flow-child-v1",
				"Source": "custom",
			}},
		}, nil).Once()

		got, err := executor.activeFlowIsPackOwned(context.Background(), "ChildFlow", nil)
		assert.NoError(t, err)
		assert.False(t, got)
		query.AssertExpectations(t)
	})

	t.Run("standard active flow source is pack-owned", func(t *testing.T) {
		query := new(MockQueryService)
		executor := &FlowExecutor{query: query}
		query.On("Query",
			mock.Anything,
			mock.MatchedBy(func(req api.QueryRequest) bool {
				return req.ObjectAPIName == constants.TableFlowDefinition
			}),
			mock.Anything,
		).Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{{
				"DeveloperName":   "ChildFlow",
				"ActiveVersionId": "flow-child-v1",
				"IsCustomized":    false,
			}},
		}, nil).Once()
		query.On("Query",
			mock.Anything,
			mock.MatchedBy(func(req api.QueryRequest) bool {
				return req.ObjectAPIName == constants.TableFlow
			}),
			mock.Anything,
		).Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{{
				"Id":     "flow-child-v1",
				"Source": flowSourceStandard,
			}},
		}, nil).Once()

		got, err := executor.activeFlowIsPackOwned(context.Background(), "ChildFlow", nil)
		assert.NoError(t, err)
		assert.True(t, got)
		query.AssertExpectations(t)
	})
}

// ---------------------------------------------------------------------------
// MockActionService for executor tests
// ---------------------------------------------------------------------------

type MockActionServiceForExec struct {
	mock.Mock
}

func (m *MockActionServiceForExec) ExecuteActionDirect(ctx context.Context, action *models.ActionMetadata, record models.SObject, user *models.UserSession) error {
	args := m.Called(ctx, action, record, user)
	return args.Error(0)
}

// ---------------------------------------------------------------------------
// MockMetadata for executor tests (wraps a minimal MetadataService interface)
// ---------------------------------------------------------------------------

// We can't mock MetadataService directly because it's a concrete struct,
// so we build a FlowExecutor with nil metadata and test the logic where
// actionMeta is nil (action-not-found paths).

// ---------------------------------------------------------------------------
// executeActionCall tests
// ---------------------------------------------------------------------------

func TestExecuteActionCall_ActionNotFound_WithFaultConnector(t *testing.T) {
	executor := &FlowExecutor{
		metadata: nil, // Will cause GetActionByID to panic if called
	}

	// We'll test the path where actionMeta is nil and fault connector is set
	// To avoid calling metadata (which is nil), we need a different approach.
	// Instead, let's test with a constructed scenario that covers the code.

	// Build interview and action call with fault connector
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	// The executeActionCall method calls e.metadata.GetActionByID - we can't
	// test this without a real MetadataService. Instead, test executeSubflow
	// and executeRecordLookup which have self-contained logic.
	_ = executor
	_ = interview
}

// ---------------------------------------------------------------------------
// executeRecordLookup tests
// ---------------------------------------------------------------------------

func TestExecuteRecordLookup_FirstRecordOnly(t *testing.T) {
	mockPersist := new(MockPersistence)
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		persistence: mockPersist,
		query:       mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	lookup := &models.FlowRecordLookup{
		Name:               "LookupAccount",
		Object:             "Account",
		GetFirstRecordOnly: true,
		Filters: []models.FlowRecordFilter{
			{Field: "Name", Operator: "=", Value: "Acme"},
		},
		OutputAssignments: []models.FlowOutputAssignment{
			{Field: "Name", AssignToReference: "accountName"},
		},
		Connector: &models.FlowConnector{TargetReference: "NextStep"},
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", mock.Anything, mock.Anything, "", "", 1).
		Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{
				{"Id": "acc-001", "Name": "Acme Corp"},
			},
		}, nil)

	next, err := executor.executeRecordLookup(context.Background(), interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "NextStep", next)
	// Verify the record was stored
	storedRecord, ok := interview.ContextData["LookupAccount"].(models.SObject)
	assert.True(t, ok)
	assert.Equal(t, "acc-001", storedRecord["Id"])
	// Verify output assignments
	assert.Equal(t, "Acme Corp", interview.ContextData["accountName"])
	mockQuery.AssertExpectations(t)
}

func TestExecuteRecordLookup_MultipleRecords(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		query: mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	lookup := &models.FlowRecordLookup{
		Name:               "LookupAccounts",
		Object:             "Account",
		GetFirstRecordOnly: false,
		Connector:          &models.FlowConnector{TargetReference: "Next"},
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", mock.Anything, mock.Anything, "", "", 200).
		Return(&models.QueryResult{
			TotalSize: 2,
			Records: []models.SObject{
				{"Id": "acc-001", "Name": "Acme"},
				{"Id": "acc-002", "Name": "Globex"},
			},
		}, nil)

	next, err := executor.executeRecordLookup(context.Background(), interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
	// Verify multiple records stored
	records, ok := interview.ContextData["LookupAccounts"].([]models.SObject)
	assert.True(t, ok)
	assert.Len(t, records, 2)
	mockQuery.AssertExpectations(t)
}

func TestExecuteRecordLookup_ForwardsSortParameters(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		query: mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	lookup := &models.FlowRecordLookup{
		Name:               "LookupLatestStatus",
		Object:             "CampaignMemberStatus",
		GetFirstRecordOnly: true,
		SortField:          "SortOrder",
		SortOrder:          "DESC",
		Connector:          &models.FlowConnector{TargetReference: "Next"},
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "CampaignMemberStatus", mock.Anything, mock.Anything, "SortOrder", "DESC", 1).
		Return(&models.QueryResult{
			TotalSize: 1,
			Records: []models.SObject{
				{"Id": "cms-004", "SortOrder": float64(4)},
			},
		}, nil)

	next, err := executor.executeRecordLookup(context.Background(), interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
	record, ok := interview.ContextData["LookupLatestStatus"].(models.SObject)
	assert.True(t, ok)
	assert.Equal(t, float64(4), record["SortOrder"])
	mockQuery.AssertExpectations(t)
}

func TestExecuteRecordLookup_ErrorWithFaultConnector(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		query: mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	lookup := &models.FlowRecordLookup{
		Name:           "LookupFailing",
		Object:         "Account",
		FaultConnector: &models.FlowConnector{TargetReference: "ErrorPath"},
		Connector:      &models.FlowConnector{TargetReference: "Success"},
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", mock.Anything, mock.Anything, "", "", 200).
		Return((*models.QueryResult)(nil), errors.New("query failed"))

	next, err := executor.executeRecordLookup(context.Background(), interview, lookup, nil)

	assert.NoError(t, err) // Error handled via fault connector
	assert.Equal(t, "ErrorPath", next)
	assert.Equal(t, "query failed", interview.ContextData["$Flow.FaultMessage"])
	mockQuery.AssertExpectations(t)
}

func TestExecuteRecordLookup_ErrorWithoutFaultConnector(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		query: mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	lookup := &models.FlowRecordLookup{
		Name:   "LookupFailing",
		Object: "Account",
		// No fault connector
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", mock.Anything, mock.Anything, "", "", 200).
		Return((*models.QueryResult)(nil), errors.New("query failed"))

	_, err := executor.executeRecordLookup(context.Background(), interview, lookup, nil)

	assert.Error(t, err)
	assert.Contains(t, err.Error(), "query failed")
	mockQuery.AssertExpectations(t)
}

func TestExecuteRecordLookup_NoConnector(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		query: mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	lookup := &models.FlowRecordLookup{
		Name:               "LookupNoConn",
		Object:             "Account",
		GetFirstRecordOnly: true,
		// No connector
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", mock.Anything, mock.Anything, "", "", 1).
		Return(&models.QueryResult{
			TotalSize: 0,
			Records:   []models.SObject{},
		}, nil)

	next, err := executor.executeRecordLookup(context.Background(), interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "", next)
	mockQuery.AssertExpectations(t)
}

func TestExecuteActionCall_EmailSimple_UsesInputParametersAndRecordContext(t *testing.T) {
	mockPersist := new(MockPersistenceForAction)
	mockEmail := new(MockEmailSender)
	logger := logging.New(context.Background())
	actionSvc := NewActionService(nil, mockPersist, nil, nil, mockEmail, logger)
	executor := &FlowExecutor{
		actions: actionSvc,
	}

	ctx := context.Background()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": map[string]interface{}{"Name": "Acme Corp"},
		},
	}
	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{Object: "Account"},
	}
	action := &models.FlowActionCall{
		Name:       "Send_Email",
		ActionName: "sendEmail",
		ActionType: "emailSimple",
		InputParameters: []models.FlowActionParameter{
			{Name: "emailTemplate", Value: "tmpl-1"},
			{Name: "recipient", Value: "recipient@test.com"},
		},
		Connector: &models.FlowConnector{TargetReference: "NextStep"},
	}

	mockPersist.On("GetRaw", ctx, constants.TableEmailTemplate, "tmpl-1").Return(models.SObject{
		"Id":        "tmpl-1",
		"Subject":   "Hello {!Record.Name}",
		"Body":      "Plain {!Record.Name}",
		"HtmlValue": "<p>HTML {!Record.Name}</p>",
		"IsActive":  int64(1),
	}, nil)
	mockEmail.On("Send", ctx, mock.MatchedBy(func(msg *ports.EmailMessage) bool {
		return len(msg.To) == 1 &&
			msg.To[0] == "recipient@test.com" &&
			msg.Subject == "Hello Acme Corp" &&
			msg.Body == "Plain Acme Corp" &&
			msg.HTML == "<p>HTML Acme Corp</p>"
	})).Return(nil)

	next, err := executor.executeActionCall(ctx, interview, flowMeta, action, nil)

	assert.NoError(t, err)
	assert.Equal(t, "NextStep", next)
	mockPersist.AssertExpectations(t)
	mockEmail.AssertExpectations(t)
}

// ---------------------------------------------------------------------------
// executeElement — routing to RecordLookup, ActionCall, Subflow
// ---------------------------------------------------------------------------

func TestExecuteElement_FindsRecordLookup(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{
		query: mockQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}
	flowMeta := &models.FlowMetadata{
		RecordLookups: []models.FlowRecordLookup{
			{
				Name:               "MyLookup",
				Object:             "Account",
				GetFirstRecordOnly: true,
				Connector:          &models.FlowConnector{TargetReference: "AfterLookup"},
			},
		},
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", mock.Anything, mock.Anything, "", "", 1).
		Return(&models.QueryResult{TotalSize: 0, Records: []models.SObject{}}, nil)

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "MyLookup", nil)
	assert.NoError(t, err)
	assert.Equal(t, "AfterLookup", next)
	mockQuery.AssertExpectations(t)
}

func TestExecuteElement_FindsRecordUpdate(t *testing.T) {
	mockPersist := new(MockPersistence)
	executor := &FlowExecutor{persistence: mockPersist}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"recId": "rec-001",
		},
	}
	flowMeta := &models.FlowMetadata{
		RecordUpdates: []models.FlowRecordUpdate{
			{
				Name:           "UpdateRec",
				Object:         "Account",
				InputReference: "recId",
				Connector:      &models.FlowConnector{TargetReference: "AfterUpdate"},
			},
		},
	}

	mockPersist.On("Update", mock.Anything, "Account", "rec-001", mock.Anything, mock.Anything).Return(nil)

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "UpdateRec", nil)
	assert.NoError(t, err)
	assert.Equal(t, "AfterUpdate", next)
	mockPersist.AssertExpectations(t)
}

func TestExecuteElement_FindsRecordDelete(t *testing.T) {
	mockPersist := new(MockPersistence)
	executor := &FlowExecutor{persistence: mockPersist}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"rec": models.SObject{"Id": "rec-001"},
		},
	}
	flowMeta := &models.FlowMetadata{
		RecordDeletes: []models.FlowRecordDelete{
			{
				Name:           "DeleteRec",
				Object:         "Account",
				InputReference: "rec",
				Connector:      &models.FlowConnector{TargetReference: "AfterDelete"},
			},
		},
	}

	mockPersist.On("Delete", mock.Anything, "Account", "rec-001", mock.Anything).Return(nil)

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "DeleteRec", nil)
	assert.NoError(t, err)
	assert.Equal(t, "AfterDelete", next)
	mockPersist.AssertExpectations(t)
}

// ---------------------------------------------------------------------------
// executeFlowMetadata — resume from paused element
// ---------------------------------------------------------------------------

func TestExecuteFlowMetadata_ResumeFromPausedElement(t *testing.T) {
	mockPersist := new(MockPersistence)
	interviewSvc := &FlowInterviewService{persistence: mockPersist}
	executor := &FlowExecutor{interviews: interviewSvc}

	// Build a flow with an assignment
	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "Step1"},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "Step2",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "result", Operator: "Assign", Value: "resumed"},
				},
				// No connector = ends flow
			},
		},
	}

	// Interview already paused at Step2
	interview := &models.FlowInterview{
		Id:             "int-001",
		CurrentElement: "Step2",
		ContextData:    map[string]interface{}{},
	}

	mockPersist.On("Update", mock.Anything, "FlowInterview", "int-001", mock.Anything, mock.Anything).
		Return(nil)

	outputs, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)
	assert.NoError(t, err)
	assert.Equal(t, "resumed", interview.ContextData["result"])
	_ = outputs
}

// ---------------------------------------------------------------------------
// executeRecordLookup — BeforeSave DetachTx guard (#1781)
// ---------------------------------------------------------------------------

// queryCtxCapture is a MockQueryService variant that records the context
// passed to QueryWithFilter so tests can assert it was the detached context.
type queryCtxCapture struct {
	MockQueryService
	capturedCtx context.Context
}

func (q *queryCtxCapture) QueryWithFilter(ctx context.Context, objectName string, filterExpr string, user *models.UserSession, sortField string, sortDir string, limit int) (*models.QueryResult, error) {
	q.capturedCtx = ctx
	args := q.MockQueryService.Called(ctx, objectName, filterExpr, user, sortField, sortDir, limit)
	if args.Get(0) == nil {
		return nil, args.Error(1)
	}
	return args.Get(0).(*models.QueryResult), args.Error(1)
}

func TestExecuteRecordLookup_BeforeSave_DetachesTx(t *testing.T) {
	// In BeforeSave context, executeRecordLookup must call DetachTx before
	// QueryWithFilter so the cross-object query runs on a fresh connection
	// with a clean TiDB read snapshot rather than the outer trigger transaction
	// (which may produce 0 rows due to REPEATABLE READ snapshot isolation). #1781
	t.Run("BeforeSave context calls DetachTx before QueryWithFilter", func(t *testing.T) {
		mockPersist := newDetachTrackingPersistence()
		captureQuery := &queryCtxCapture{}
		executor := &FlowExecutor{
			persistence: mockPersist,
			query:       captureQuery,
		}

		interview := &models.FlowInterview{
			ContextData: map[string]interface{}{
				"$Record": models.SObject{constants.FieldID: "opp-001"},
			},
		}
		lookup := &models.FlowRecordLookup{
			Name:               "get_products",
			Object:             "Opportunity_Product__c",
			GetFirstRecordOnly: true,
			Connector:          &models.FlowConnector{TargetReference: "Next"},
		}

		captureQuery.MockQueryService.On("QueryWithFilter",
			mock.Anything, "Opportunity_Product__c", mock.Anything, mock.Anything, "", "", 1).
			Return(&models.QueryResult{
				TotalSize: 1,
				Records:   []models.SObject{{"Id": "prod-001"}},
			}, nil)

		ctx := WithBeforeSaveFlow(WithTransientFlowInterviews(context.Background()))
		next, err := executor.executeRecordLookup(ctx, interview, lookup, nil)

		assert.NoError(t, err)
		assert.Equal(t, "Next", next)
		assert.True(t, mockPersist.detachCalled,
			"DetachTx must be called in BeforeSave context before cross-object GetRecords (#1781)")
		// The context passed to QueryWithFilter must carry the detach sentinel.
		assert.NotNil(t, captureQuery.capturedCtx.Value(detachSentinelKey{}),
			"QueryWithFilter must receive the detached context, not the outer transaction context")
		captureQuery.MockQueryService.AssertExpectations(t)
	})

	t.Run("AfterSave context does NOT call DetachTx for GetRecords", func(t *testing.T) {
		// In AfterSave context there is no open BeforeSave transaction; DetachTx
		// is only needed for BeforeSave. This test guards against over-detaching.
		mockPersist := newDetachTrackingPersistence()
		mockQuery := new(MockQueryService)
		executor := &FlowExecutor{
			persistence: mockPersist,
			query:       mockQuery,
		}

		interview := &models.FlowInterview{
			ContextData: map[string]interface{}{},
		}
		lookup := &models.FlowRecordLookup{
			Name:               "get_products",
			Object:             "Opportunity_Product__c",
			GetFirstRecordOnly: true,
			Connector:          &models.FlowConnector{TargetReference: "Next"},
		}

		mockQuery.On("QueryWithFilter",
			mock.Anything, "Opportunity_Product__c", mock.Anything, mock.Anything, "", "", 1).
			Return(&models.QueryResult{
				TotalSize: 0,
				Records:   []models.SObject{},
			}, nil)

		// No WithBeforeSaveFlow — simulates AfterSave listener context.
		ctx := WithTransientFlowInterviews(context.Background())
		next, err := executor.executeRecordLookup(ctx, interview, lookup, nil)

		assert.NoError(t, err)
		assert.Equal(t, "Next", next)
		assert.False(t, mockPersist.detachCalled,
			"DetachTx must NOT be called in AfterSave context for GetRecords")
		mockQuery.AssertExpectations(t)
	})
}

// ---------------------------------------------------------------------------
// executeRecordLookup — BeforeSave cross-object vs same-object DetachTx guard
//
// HIGH H1: Same-object lookups in BeforeSave must NOT DetachTx so they can
// see the outer transaction's uncommitted $Record mutations. Cross-object
// lookups must DetachTx to escape TiDB REPEATABLE READ snapshot staleness.
// ---------------------------------------------------------------------------

func TestExecuteRecordLookup_BeforeSaveCrossObjectUsesDetachTx(t *testing.T) {
	// Cross-object: $Record is Opportunity, lookup is Opportunity_Product__c.
	// DetachTx must be called so the query sees rows committed by other sessions.
	mockPersist := newDetachTrackingPersistence()
	captureQuery := &queryCtxCapture{}
	executor := &FlowExecutor{
		persistence: mockPersist,
		query:       captureQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": models.SObject{
				constants.FieldID: "opp-cross-001",
				"attributes":      map[string]interface{}{"type": "Opportunity"},
			},
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "get_products",
		Object:             "Opportunity_Product__c",
		GetFirstRecordOnly: true,
		Connector:          &models.FlowConnector{TargetReference: "Next"},
	}

	captureQuery.MockQueryService.On("QueryWithFilter",
		mock.Anything, "Opportunity_Product__c", mock.Anything, mock.Anything, "", "", 1).
		Return(&models.QueryResult{
			TotalSize: 1,
			Records:   []models.SObject{{"Id": "prod-001"}},
		}, nil)

	ctx := WithBeforeSaveFlow(WithTransientFlowInterviews(context.Background()))
	next, err := executor.executeRecordLookup(ctx, interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
	assert.True(t, mockPersist.detachCalled,
		"DetachTx must be called for cross-object BeforeSave GetRecords (#1781)")
	assert.NotNil(t, captureQuery.capturedCtx.Value(detachSentinelKey{}),
		"QueryWithFilter must receive the detached context for cross-object lookup")
	captureQuery.MockQueryService.AssertExpectations(t)
}

func TestExecuteRecordLookup_BeforeSaveSameObjectPreservesUncommitted(t *testing.T) {
	// Same-object: $Record is Opportunity, lookup is also Opportunity.
	// DetachTx must NOT be called so the query sees the outer transaction's
	// uncommitted $Record mutations (BeforeSave in-memory semantics).
	mockPersist := newDetachTrackingPersistence()
	captureQuery := &queryCtxCapture{}
	executor := &FlowExecutor{
		persistence: mockPersist,
		query:       captureQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": models.SObject{
				constants.FieldID: "opp-same-001",
				"attributes":      map[string]interface{}{"type": "Opportunity"},
			},
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "get_opportunity",
		Object:             "Opportunity", // same object as trigger
		GetFirstRecordOnly: true,
		Connector:          &models.FlowConnector{TargetReference: "Next"},
	}

	captureQuery.MockQueryService.On("QueryWithFilter",
		mock.Anything, "Opportunity", mock.Anything, mock.Anything, "", "", 1).
		Return(&models.QueryResult{
			TotalSize: 1,
			Records:   []models.SObject{{"Id": "opp-same-001"}},
		}, nil)

	ctx := WithBeforeSaveFlow(WithTransientFlowInterviews(context.Background()))
	next, err := executor.executeRecordLookup(ctx, interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
	assert.False(t, mockPersist.detachCalled,
		"DetachTx must NOT be called for same-object BeforeSave GetRecords — outer tx must see uncommitted $Record mutations")
	// Context passed to QueryWithFilter must NOT carry the detach sentinel.
	assert.Nil(t, captureQuery.capturedCtx.Value(detachSentinelKey{}),
		"QueryWithFilter must receive the original (non-detached) context for same-object lookup")
	captureQuery.MockQueryService.AssertExpectations(t)
}

// ---------------------------------------------------------------------------
// executeRecordLookup — atomic-flow boundary keeps cross-object lookup attached
//
// Issue #2098: an atomic parent flow (e.g. Std_ConvertLead) creates a new
// Opportunity in the same transaction, then triggers a child BeforeSave Update
// flow on Lead via stampConvertedAnchors. The child flow's cross-object
// recordLookup against the just-created Opportunity must see the parent's
// uncommitted INSERT — read-your-own-writes inside the same tx. Detaching here
// (the #1781 fix for stale snapshots from external sessions) regressed this
// path because a fresh connection cannot see uncommitted writes from another tx.
//
// The discriminator is the atomic-flow boundary: when the outer tx is one
// (RunAtomically=true), keep the lookup attached so the same-tx INSERT is
// visible. Outside an atomic boundary, the #1781 detach behavior is preserved.
// ---------------------------------------------------------------------------

func TestExecuteRecordLookup_BeforeSaveCrossObjectInsideAtomicFlowDoesNotDetach(t *testing.T) {
	// Cross-object: $Record is Lead, lookup is Opportunity (just inserted by
	// the atomic parent flow earlier in the same transaction).
	mockPersist := newDetachTrackingPersistence()
	captureQuery := &queryCtxCapture{}
	executor := &FlowExecutor{
		persistence: mockPersist,
		query:       captureQuery,
	}

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": models.SObject{
				constants.FieldID:        "lead-2098-001",
				"attributes":             map[string]interface{}{"type": "Lead"},
				"IsConverted":            true,
				"ConvertedOpportunityId": "opp-2098-001",
			},
		},
	}
	lookup := &models.FlowRecordLookup{
		Name:               "getopportunity",
		Object:             "Opportunity",
		GetFirstRecordOnly: true,
		Connector:          &models.FlowConnector{TargetReference: "Next"},
	}

	captureQuery.MockQueryService.On("QueryWithFilter",
		mock.Anything, "Opportunity", mock.Anything, mock.Anything, "", "", 1).
		Return(&models.QueryResult{
			TotalSize: 1,
			Records:   []models.SObject{{"Id": "opp-2098-001", "Name": "Test Opp"}},
		}, nil)

	// BeforeSave + atomic-flow boundary (parent flow ran with runAtomically=true).
	ctx := WithBeforeSaveFlow(withAtomicFlowBoundary(WithTransientFlowInterviews(context.Background())))
	next, err := executor.executeRecordLookup(ctx, interview, lookup, nil)

	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
	assert.False(t, mockPersist.detachCalled,
		"DetachTx must NOT be called when BeforeSave runs inside an atomic-flow boundary — read-your-own-writes from the same tx is required (#2098)")
	assert.Nil(t, captureQuery.capturedCtx.Value(detachSentinelKey{}),
		"QueryWithFilter must receive the original (non-detached) context so it sees the atomic parent's uncommitted INSERT")
	captureQuery.MockQueryService.AssertExpectations(t)
}