package automation

import (
	"context"
	"testing"

	"github.com/nexuscrm/backend/pkg/logging"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
)

func TestExecuteElement_NotFound(t *testing.T) {
	executor := newTestExecutor()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}
	flowMeta := &models.FlowMetadata{}
	user := &models.UserSession{ID: "user-1"}

	_, err := executor.executeElement(context.Background(), interview, flowMeta, "NonExistent", user)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "flow element")
}

func TestExecuteElement_FindsDecision(t *testing.T) {
	executor := newTestExecutor()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{"status": "Active"},
	}
	flowMeta := &models.FlowMetadata{
		Decisions: []models.FlowDecision{
			{
				Name: "CheckStatus",
				Rules: []models.FlowDecisionRule{
					{
						Name:           "isActive",
						ConditionLogic: "and",
						Conditions: []models.FlowCondition{
							{LeftValueReference: "status", Operator: "EqualTo", RightValue: "Active"},
						},
						Connector: &models.FlowConnector{TargetReference: "ActivePath"},
					},
				},
			},
		},
	}
	user := &models.UserSession{ID: "user-1"}

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "CheckStatus", user)
	assert.NoError(t, err)
	assert.Equal(t, "ActivePath", next)
}

func TestExecuteElement_FindsAssignment(t *testing.T) {
	executor := newTestExecutor()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}
	flowMeta := &models.FlowMetadata{
		Assignments: []models.FlowAssignment{
			{
				Name: "SetVar",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "myVar", Operator: "Assign", Value: "test"},
				},
				Connector: &models.FlowConnector{TargetReference: "Next"},
			},
		},
	}
	user := &models.UserSession{ID: "user-1"}

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "SetVar", user)
	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
	assert.Equal(t, "test", interview.ContextData["myVar"])
}

func TestExecuteElement_FindsLoop(t *testing.T) {
	executor := newTestExecutor()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"items": []interface{}{"a"},
		},
	}
	flowMeta := &models.FlowMetadata{
		Loops: []models.FlowLoop{
			{
				Name:                       "MyLoop",
				CollectionReference:        "items",
				AssignNextValueToReference: "current",
				NextValueConnector:         &models.FlowConnector{TargetReference: "LoopBody"},
				NoMoreValuesConnector:      &models.FlowConnector{TargetReference: "AfterLoop"},
			},
		},
	}
	user := &models.UserSession{ID: "user-1"}

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "MyLoop", user)
	assert.NoError(t, err)
	assert.Equal(t, "LoopBody", next)
	assert.Equal(t, "a", interview.ContextData["current"])
}

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

	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}
	flowMeta := &models.FlowMetadata{
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:      "CreateRec",
				Object:    "Account",
				Connector: &models.FlowConnector{TargetReference: "Next"},
			},
		},
	}

	mockPersist.On("Insert", context.Background(), "Account", models.SObject{}, (*models.UserSession)(nil)).
		Return(models.SObject{"Id": "acc-001"}, nil)

	next, err := executor.executeElement(context.Background(), interview, flowMeta, "CreateRec", nil)
	assert.NoError(t, err)
	assert.Equal(t, "Next", next)
}

func TestExecuteFlowMetadata_NoStartElement(t *testing.T) {
	executor := newTestExecutor()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	// No start element
	flowMeta := &models.FlowMetadata{}
	_, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "no start element")

	// Start without connector
	flowMeta2 := &models.FlowMetadata{Start: &models.FlowStart{}}
	_, err = executor.executeFlowMetadata(context.Background(), interview, flowMeta2, nil)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "no start element")
}

// TestExecuteFlowMetadata_ScheduledPathFallback verifies that flows with no
// Start.Connector (AsyncAfterCommit / scheduled-path flows) can still enter
// the graph via Start.ScheduledPaths[].Connector. Regression for #1790.
func TestExecuteFlowMetadata_ScheduledPathFallback(t *testing.T) {
	executor := newTestExecutor()
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{},
	}

	// Flow where Start.Connector is nil but ScheduledPaths holds the entry.
	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{
			// Connector intentionally nil — mirrors an AsyncAfterCommit flow.
			ScheduledPaths: []models.FlowScheduledPath{
				{
					Name:      "AfterOneMin",
					Connector: &models.FlowConnector{TargetReference: "SetVar"},
				},
			},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "SetVar",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "result", Operator: "Assign", Value: "async-ran"},
				},
				// No connector — flow ends here.
			},
		},
		Variables: []models.FlowVariable{
			{Name: "result", IsOutput: true},
		},
	}

	mockPersist := new(MockPersistence)
	executor.interviews = &FlowInterviewService{persistence: mockPersist}

	outputs, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)
	assert.NoError(t, err)
	assert.Equal(t, "async-ran", outputs["result"])

	// A scheduled path with an empty connector still returns no-start-element.
	flowMetaEmpty := &models.FlowMetadata{
		Start: &models.FlowStart{
			ScheduledPaths: []models.FlowScheduledPath{
				{Name: "NoConnector"},
			},
		},
	}
	_, err = executor.executeFlowMetadata(context.Background(), interview, flowMetaEmpty, nil)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "no start element")
}

func TestExecuteFlowMetadata_CollectsOutputVariables(t *testing.T) {
	executor := newTestExecutor()

	// Build a flow that sets a variable and outputs it
	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "SetOutput"},
		},
		Variables: []models.FlowVariable{
			{Name: "outputVar", IsOutput: true},
			{Name: "internalVar", IsOutput: false},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "SetOutput",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "outputVar", Operator: "Assign", Value: "result-value"},
					{AssignToReference: "internalVar", Operator: "Assign", Value: "internal-value"},
				},
				// No connector = flow ends
			},
		},
	}

	mockPersist := new(MockPersistence)
	interviewSvc := &FlowInterviewService{persistence: mockPersist}
	executor.interviews = interviewSvc

	interview := &models.FlowInterview{
		Id:          "int-001",
		ContextData: map[string]interface{}{},
	}

	outputs, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)
	assert.NoError(t, err)
	assert.Equal(t, "result-value", outputs["outputVar"])
	// Internal variable should not be in outputs
	_, exists := outputs["internalVar"]
	assert.False(t, exists)
	mockPersist.AssertNotCalled(t, "Update", mock.Anything, "FlowInterview", "int-001", mock.Anything, mock.Anything)
}

func TestExecuteFlowMetadata_IncludesRuntimeRecordOutputs(t *testing.T) {
	executor := newTestExecutor()

	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{
			TriggerType: "RecordBeforeSave",
			Connector:   &models.FlowConnector{TargetReference: "SetStatus"},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "SetStatus",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "$Record.Account_Status__c", Operator: "Assign", Value: "new_user"},
				},
			},
		},
	}

	interview := &models.FlowInterview{
		Id: "int-001",
		ContextData: map[string]interface{}{
			"$Record": models.SObject{
				"Portal_ID__c":      "portal-1",
				"Account_Status__c": nil,
			},
		},
	}

	outputs, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)
	assert.NoError(t, err)
	record, ok := outputs["$Record"].(models.SObject)
	assert.True(t, ok)
	assert.Equal(t, "new_user", record["Account_Status__c"])
}

func TestExecuteFlowMetadata_DoesNotPersistPerElementCheckpoints(t *testing.T) {
	executor := newTestExecutor()

	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "Step1"},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "Step1",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "step1", Operator: "Assign", Value: "done"},
				},
				Connector: &models.FlowConnector{TargetReference: "Step2"},
			},
			{
				Name: "Step2",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "step2", Operator: "Assign", Value: "done"},
				},
			},
		},
	}

	mockPersist := new(MockPersistence)
	executor.interviews = &FlowInterviewService{persistence: mockPersist}

	interview := &models.FlowInterview{
		Id:          "int-001",
		ContextData: map[string]interface{}{},
	}

	outputs, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)

	assert.NoError(t, err)
	assert.Empty(t, outputs)
	assert.Equal(t, "done", interview.ContextData["step1"])
	assert.Equal(t, "done", interview.ContextData["step2"])
	mockPersist.AssertNotCalled(t, "Update", mock.Anything, "FlowInterview", "int-001", mock.Anything, mock.Anything)
}

// TestExecuteFlowMetadata_BeforeSaveMutatesSharedRecord verifies the full
// before-save flow contract: an assignment to `$Record.FieldName` must mutate
// the caller's in-flight sObject, since persistence_service relies on the
// shared map reference to persist the new value during Insert/Update.
//
// Regression for ywang-official/MetaDrivenPlatform4AI#550 — the
// `Lead_Assignment_Rule_modification_for_NA_EMEA` flow was firing on Lead
// create/update but OwnerId was never being set, because dotted assignments
// wrote to a flat ContextData key instead of mutating the $Record sub-map.
func TestExecuteFlowMetadata_BeforeSaveMutatesSharedRecord(t *testing.T) {
	executor := newTestExecutor()

	// This is the shared map reference the caller holds — in production it is
	// `data` inside persistence_insert.go. The flow engine must mutate this
	// very map so the subsequent repo.Insert picks up the new OwnerId.
	record := models.SObject{
		"Id":        "lead-1",
		"Region__c": "NA",
		"OwnerId":   nil,
	}

	flowMeta := &models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "AssignOwner"},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "AssignOwner",
				Items: []models.FlowAssignmentItem{
					{
						AssignToReference: "$Record.OwnerId",
						Operator:          "Assign",
						Value:             "005RC00000BN0iXYAT",
					},
				},
			},
		},
	}

	executor.interviews = &FlowInterviewService{persistence: new(MockPersistence)}
	interview := &models.FlowInterview{
		Id: "int-550",
		// Mirrors initializeFlowContext after inputs["$Record"] = record.
		ContextData: map[string]interface{}{
			"$Record": record,
		},
	}

	_, err := executor.executeFlowMetadata(context.Background(), interview, flowMeta, nil)
	assert.NoError(t, err)

	// The caller's shared map must see the OwnerId the flow set.
	assert.Equal(t, "005RC00000BN0iXYAT", record["OwnerId"],
		"before-save flow must mutate the shared $Record map so persistence picks up the new OwnerId")
}

// assertNoFlowInterviewPersistence asserts that a mock PersistenceProvider
// received zero Insert or Update calls targeted at the FlowInterview table.
// It counts calls directly from mock.Calls rather than using testify's
// AssertNotCalled, which only fires when a matching expectation has been
// registered — a genuine regression that introduced an Upsert call or used
// a different table name would otherwise pass a weaker assertion silently.
func assertNoFlowInterviewPersistence(t *testing.T, m *MockPersistence) {
	t.Helper()
	for _, call := range m.Calls {
		if call.Method != "Insert" && call.Method != "Update" {
			continue
		}
		if len(call.Arguments) < 2 {
			continue
		}
		if obj, ok := call.Arguments[1].(string); ok && obj == "FlowInterview" {
			t.Errorf("unexpected FlowInterview persistence call: %s(%v)", call.Method, call.Arguments)
		}
	}
}

// newTransientExecutor builds a FlowExecutor with a mock persistence layer,
// a real FlowInterviewService wrapping it, and a logger safe for tests.
func newTransientExecutor(mockPersist *MockPersistence) *FlowExecutor {
	return &FlowExecutor{
		persistence: mockPersist,
		interviews:  &FlowInterviewService{persistence: mockPersist},
		logger:      logging.New(context.Background()),
	}
}

// TestExecuteFlowVersion_TransientInterview_NoPersistence verifies that a
// synchronous record-triggered run — one whose context was marked with
// WithTransientFlowInterviews — finishes without hitting the FlowInterview
// table at all. Each persisted interview costs ~4 seconds on remote TiDB
// Cloud and a Lead with ~24 before-save flows would otherwise take ~160s.
//
// This is the perf fix for the save-latency problem observed when fixing
// ywang-official/MetaDrivenPlatform4AI#550.
func TestExecuteFlowVersion_TransientInterview_NoPersistence(t *testing.T) {
	mockPersist := new(MockPersistence)
	executor := newTransientExecutor(mockPersist)

	flowVersion := &models.FlowVersion{
		Id: "301X-test",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "SetOwner"},
			},
			Assignments: []models.FlowAssignment{
				{
					Name: "SetOwner",
					Items: []models.FlowAssignmentItem{
						{
							AssignToReference: "$Record.OwnerId",
							Operator:          "Assign",
							Value:             "005x",
						},
					},
				},
			},
		},
	}

	inputs := map[string]interface{}{
		"$Record": models.SObject{"Id": "lead-1", "Region__c": "NA"},
	}
	user := &models.UserSession{ID: "user-1"}

	ctx := WithTransientFlowInterviews(context.Background())
	result, err := executor.executeFlowVersion(ctx, flowVersion, "test_flow", inputs, user)
	assert.NoError(t, err)
	assert.True(t, result.IsSuccess)

	assertNoFlowInterviewPersistence(t, mockPersist)
}

// TestExecuteFlowVersion_TransientInterview_EntrySkipped verifies that even
// when a record fails the entry conditions (and execution short-circuits),
// no FlowInterview rows are written.
func TestExecuteFlowVersion_TransientInterview_EntrySkipped(t *testing.T) {
	mockPersist := new(MockPersistence)
	executor := newTransientExecutor(mockPersist)

	flowVersion := &models.FlowVersion{
		Id: "301X-skip",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "Noop"},
				Filters: []models.FlowRecordFilter{
					{Field: "Region__c", Operator: "EqualTo", Value: "NA"},
				},
			},
		},
	}

	// Record has Region=EMEA, flow filter requires NA → entry should skip.
	inputs := map[string]interface{}{
		"$Record": models.SObject{"Id": "lead-2", "Region__c": "EMEA"},
	}

	ctx := WithTransientFlowInterviews(context.Background())
	result, err := executor.executeFlowVersion(ctx, flowVersion, "test_flow_skip", inputs, &models.UserSession{ID: "user-1"})
	assert.NoError(t, err)
	assert.True(t, result.IsSuccess)
	assert.Equal(t, true, result.OutputValues["_skipped"])

	assertNoFlowInterviewPersistence(t, mockPersist)
}

// TestExecuteFlowVersion_TransientInterview_ErrorPath pins the tradeoff
// documented in the perf fix: when a transient flow run errors mid-
// execution, no FlowInterview row is written (the error is only visible
// via the structured logger). A future refactor that reintroduces
// persistence on the error path without this test's blessing would be
// caught here.
func TestExecuteFlowVersion_TransientInterview_ErrorPath(t *testing.T) {
	mockPersist := new(MockPersistence)
	executor := newTransientExecutor(mockPersist)

	// The Start element references an undefined target, so executeElement
	// will return an error when it tries to resolve it — exercising the
	// err path inside executeFlowVersion.
	flowVersion := &models.FlowVersion{
		Id: "301X-err",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "NonExistentElement"},
			},
		},
	}

	inputs := map[string]interface{}{
		"$Record": models.SObject{"Id": "lead-3"},
	}

	ctx := WithTransientFlowInterviews(context.Background())
	result, err := executor.executeFlowVersion(ctx, flowVersion, "test_flow_err", inputs, &models.UserSession{ID: "user-1"})
	assert.NoError(t, err, "executeFlowVersion wraps flow errors into the result, it should not return an error itself")
	assert.False(t, result.IsSuccess)
	assert.NotEmpty(t, result.Errors)
	assert.Equal(t, "FLOW_ERROR", result.Errors[0].StatusCode)

	assertNoFlowInterviewPersistence(t, mockPersist)
}

// TestExecuteFlowVersion_PersistentMode_WritesInterview guards the default
// (persistent) branch — any caller that has NOT opted into transient mode
// (scheduler, REST-invoked flows, subflow chains) must still see its
// FlowInterview row inserted and completed. Without this test a future
// refactor of shouldPersistFlowInterview could silently flip the default
// and lose operator visibility for scheduled flows.
func TestExecuteFlowVersion_PersistentMode_WritesInterview(t *testing.T) {
	mockPersist := new(MockPersistence)
	executor := newTransientExecutor(mockPersist)

	mockPersist.On("Insert", mock.Anything, "FlowInterview", mock.Anything, mock.Anything).
		Return(models.SObject{"Id": "persisted-1"}, nil).Once()
	mockPersist.On("Update", mock.Anything, "FlowInterview", "persisted-1", mock.Anything, mock.Anything).
		Return(nil).Once()

	flowVersion := &models.FlowVersion{
		Id: "301X-persistent",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "Noop"},
			},
			Assignments: []models.FlowAssignment{
				{Name: "Noop"},
			},
		},
	}

	// Default context — no WithTransientFlowInterviews, so persistence must fire.
	result, err := executor.executeFlowVersion(context.Background(), flowVersion, "test_flow_persisted", map[string]interface{}{}, &models.UserSession{ID: "user-1"})
	assert.NoError(t, err)
	assert.True(t, result.IsSuccess)

	mockPersist.AssertExpectations(t)
}

func TestBuildScheduledRunInputs_NoObjectRunsOnce(t *testing.T) {
	executor := newTestExecutor()

	inputs, err := executor.buildScheduledRunInputs(context.Background(), &models.FlowVersion{
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{TriggerType: "Scheduled"},
		},
	}, nil)

	assert.NoError(t, err)
	assert.Len(t, inputs, 1)
	assert.Nil(t, inputs[0])
}

func TestBuildScheduledRunInputs_QueriesRecordsAndAppliesEntryConditions(t *testing.T) {
	mockQuery := new(MockQueryService)
	executor := &FlowExecutor{query: mockQuery}
	user := &models.UserSession{ID: "system"}

	flowVersion := &models.FlowVersion{
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Object:      "Account",
				FilterLogic: "AND",
				Filters: []models.FlowRecordFilter{
					{Field: "Name", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Acme"}},
				},
			},
		},
	}

	mockQuery.On("QueryWithFilter", mock.Anything, "Account", "Name = 'Acme'", user, "", "", scheduledFlowQueryLimit).
		Return(&models.QueryResult{
			Records: []models.SObject{
				{"Id": "001", "Name": "Acme"},
				{"Id": "002", "Name": "Globex"},
			},
		}, nil).Once()

	inputs, err := executor.buildScheduledRunInputs(context.Background(), flowVersion, user)

	assert.NoError(t, err)
	assert.Len(t, inputs, 1)
	assert.Equal(t, models.SObject{"Id": "001", "Name": "Acme", "attributes": map[string]interface{}{"type": "Account"}}, inputs[0]["$Record"])
	mockQuery.AssertExpectations(t)
}