package automation

import (
	"context"
	"strings"
	"testing"

	"github.com/nexuscrm/automation/triggersub"
	eventFeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedEvents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestValidateAtomicFlowForExecutionRejectsUnsupportedTriggerShapes(t *testing.T) {
	t.Parallel()

	triggerRootCtx, alreadyInFlight := withRecordTrigger(context.Background(), "00Q", "00Q000000000001AAA", "300000000000001AAA")
	if alreadyInFlight {
		t.Fatal("withRecordTrigger() unexpectedly reported already in flight")
	}
	triggerRootCtx = withRecordTriggerRoot(triggerRootCtx)

	tests := []struct {
		name        string
		ctx         context.Context
		processType string
		triggerType string
		object      string
		variables   []models.FlowVariable
		wantMention string
		wantErr     bool
	}{
		{name: "before-save allowed on trigger root", ctx: triggerRootCtx, triggerType: "RecordBeforeSave", object: "Lead"},
		{name: "after-save rejected on trigger root", ctx: triggerRootCtx, triggerType: "RecordAfterSave", object: "Lead", wantMention: "RecordAfterSave", wantErr: true},
		{name: "after-save alias rejected on trigger root", ctx: triggerRootCtx, triggerType: "onAllChanges", object: "Lead", wantMention: "RecordAfterSave", wantErr: true},
		{name: "after-save mixed case rejected on trigger root", ctx: triggerRootCtx, triggerType: "recordaftersave", object: "Lead", wantMention: "RecordAfterSave", wantErr: true},
		{name: "before-delete rejected on trigger root", ctx: triggerRootCtx, triggerType: "RecordBeforeDelete", object: "Lead", wantMention: "RecordBeforeDelete", wantErr: true},
		{name: "after-delete rejected on trigger root", ctx: triggerRootCtx, triggerType: "RecordAfterDelete", object: "Lead", wantMention: "RecordAfterDelete", wantErr: true},
		{name: "after-save allowed for direct invoke", ctx: context.Background(), triggerType: "RecordAfterSave", object: "Lead"},
		{name: "after-save alias allowed for direct invoke", ctx: context.Background(), triggerType: "onAllChanges", object: "Lead"},
		{
			name:        "workflow-derived after-save rejected on trigger root",
			ctx:         triggerRootCtx,
			processType: "Workflow",
			variables: []models.FlowVariable{{
				Name:       "myVariable_current",
				DataType:   "SObject",
				ObjectType: "Lead",
			}},
			wantMention: "RecordAfterSave",
			wantErr:     true,
		},
		{
			name:        "workflow-derived after-save allowed for direct invoke",
			ctx:         context.Background(),
			processType: "Workflow",
			variables: []models.FlowVariable{{
				Name:       "myVariable_current",
				DataType:   "SObject",
				ObjectType: "Lead",
			}},
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			err := validateAtomicFlowForExecution(tc.ctx, tc.processType, &models.FlowMetadata{
				RunAtomically: true,
				Start:         &models.FlowStart{TriggerType: tc.triggerType, Object: tc.object},
				Variables:     tc.variables,
			})
			if tc.wantErr {
				if err == nil {
					t.Fatal("validateAtomicFlowForExecution() error = nil, want error")
				}
				if !strings.Contains(err.Error(), tc.wantMention) {
					t.Fatalf("error = %v, want mention of %q", err, tc.wantMention)
				}
				return
			}
			if err != nil {
				t.Fatalf("validateAtomicFlowForExecution() error = %v, want nil", err)
			}
		})
	}
}

func TestValidateAtomicFlowForExecutionAllowsUnsupportedTriggerShapesForSubflows(t *testing.T) {
	t.Parallel()

	rootCtx, alreadyInFlight := withRecordTrigger(context.Background(), "00Q", "00Q000000000001AAA", "300000000000001AAA")
	if alreadyInFlight {
		t.Fatal("withRecordTrigger() unexpectedly reported already in flight")
	}
	rootCtx = withRecordTriggerRoot(rootCtx)
	subflowCtx, err := withFlowCall(rootCtx, "ChildFlow")
	if err != nil {
		t.Fatalf("withFlowCall() error = %v", err)
	}

	err = validateAtomicFlowForExecution(subflowCtx, "", &models.FlowMetadata{
		RunAtomically: true,
		Start:         &models.FlowStart{TriggerType: "RecordAfterSave", Object: "Lead"},
	})
	if err != nil {
		t.Fatalf("validateAtomicFlowForExecution() error = %v, want nil for subflow invocation", err)
	}
}

func TestValidateAtomicFlowForExecutionRejectsUnsupportedTriggerShapesForNestedTriggerRoots(t *testing.T) {
	t.Parallel()

	parentCtx, err := withFlowCall(context.Background(), "ParentFlow")
	if err != nil {
		t.Fatalf("withFlowCall() error = %v", err)
	}
	nestedRootCtx, alreadyInFlight := withRecordTrigger(parentCtx, "00Q", "00Q000000000001AAA", "300000000000001AAA")
	if alreadyInFlight {
		t.Fatal("withRecordTrigger() unexpectedly reported already in flight")
	}
	nestedRootCtx = withRecordTriggerRoot(nestedRootCtx)

	err = validateAtomicFlowForExecution(nestedRootCtx, "Workflow", &models.FlowMetadata{
		RunAtomically: true,
		Variables: []models.FlowVariable{{
			Name:       "myVariable_current",
			DataType:   "SObject",
			ObjectType: "Lead",
		}},
	})
	if err == nil {
		t.Fatal("validateAtomicFlowForExecution() error = nil, want nested trigger-root rejection")
	}
	if !strings.Contains(err.Error(), "RecordAfterSave") {
		t.Fatalf("error = %v, want mention of RecordAfterSave", err)
	}
}

func TestValidateAtomicFlowForExecutionRejectsInheritedAtomicUnsupportedTriggerRoots(t *testing.T) {
	t.Parallel()

	triggerRootCtx, alreadyInFlight := withRecordTrigger(context.Background(), "00Q", "00Q000000000001AAA", "300000000000001AAA")
	if alreadyInFlight {
		t.Fatal("withRecordTrigger() unexpectedly reported already in flight")
	}
	triggerRootCtx = withRecordTriggerRoot(triggerRootCtx)
	inheritedAtomicCtx := withAtomicFlowBoundary(triggerRootCtx)

	tests := []struct {
		name        string
		processType string
		triggerType string
		object      string
		variables   []models.FlowVariable
		wantMention string
		wantErr     bool
	}{
		{name: "before-save allowed under inherited atomic boundary", triggerType: "RecordBeforeSave", object: "Lead"},
		{name: "after-save rejected under inherited atomic boundary", triggerType: "RecordAfterSave", object: "Lead", wantMention: "RecordAfterSave", wantErr: true},
		{name: "before-delete rejected under inherited atomic boundary", triggerType: "RecordBeforeDelete", object: "Lead", wantMention: "RecordBeforeDelete", wantErr: true},
		{name: "after-delete rejected under inherited atomic boundary", triggerType: "RecordAfterDelete", object: "Lead", wantMention: "RecordAfterDelete", wantErr: true},
		{
			name:        "workflow-derived after-save rejected under inherited atomic boundary",
			processType: "Workflow",
			variables: []models.FlowVariable{{
				Name:       "myVariable_current",
				DataType:   "SObject",
				ObjectType: "Lead",
			}},
			wantMention: "RecordAfterSave",
			wantErr:     true,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			err := validateAtomicFlowForExecution(inheritedAtomicCtx, tc.processType, &models.FlowMetadata{
				RunAtomically: false,
				Start:         &models.FlowStart{TriggerType: tc.triggerType, Object: tc.object},
				Variables:     tc.variables,
			})
			if tc.wantErr {
				if err == nil {
					t.Fatal("validateAtomicFlowForExecution() error = nil, want error")
				}
				if !strings.Contains(err.Error(), tc.wantMention) {
					t.Fatalf("error = %v, want mention of %q", err, tc.wantMention)
				}
				return
			}
			if err != nil {
				t.Fatalf("validateAtomicFlowForExecution() error = %v, want nil", err)
			}
		})
	}
}

func TestValidateAtomicFlowForExecutionRejectsPushToConnectorExternalIO(t *testing.T) {
	t.Parallel()

	exec := NewFlowExecutor(nil, nil, nil, nil, nil, nil, nil, WithFlowActionCall(RuntimeOutboundIntentFlowActionCallRegistration()))
	inheritedAtomicCtx := withAtomicFlowBoundary(context.Background())
	tests := []struct {
		name string
		ctx  context.Context
		meta models.FlowMetadata
	}{
		{
			name: "direct atomic flow",
			ctx:  context.Background(),
			meta: models.FlowMetadata{
				RunAtomically: true,
				ActionCalls: []models.FlowActionCall{{
					Name:       "Push",
					ActionType: constants.ActionTypePushToConnector,
				}},
			},
		},
		{
			name: "inherited atomic boundary",
			ctx:  inheritedAtomicCtx,
			meta: models.FlowMetadata{
				ActionCalls: []models.FlowActionCall{{
					Name:       "Push",
					ActionType: constants.ActionTypePushToConnector,
				}},
			},
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			err := exec.validateAtomicFlowForExecution(tc.ctx, "", &tc.meta)
			if err == nil {
				t.Fatal("validateAtomicFlowForExecution() error = nil, want PushToConnector rejection")
			}
			if !strings.Contains(err.Error(), "PushToConnector") {
				t.Fatalf("error = %v, want PushToConnector mention", err)
			}
		})
	}
}

func TestValidateAtomicFlowForExecutionRejectsEmailIOActionCalls(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name        string
		ctx         context.Context
		runAtomic   bool
		actionType  string
		wantErr     bool
		wantMention string
	}{
		{
			name:       "non-atomic sendEmail allowed",
			ctx:        context.Background(),
			actionType: flowSendEmailActionType,
		},
		{
			name:        "direct atomic emailAlert rejected",
			ctx:         context.Background(),
			runAtomic:   true,
			actionType:  "emailAlert",
			wantErr:     true,
			wantMention: "emailAlert",
		},
		{
			name:        "direct atomic sendEmail rejected",
			ctx:         context.Background(),
			runAtomic:   true,
			actionType:  flowSendEmailActionType,
			wantErr:     true,
			wantMention: flowSendEmailActionType,
		},
		{
			name:        "direct atomic emailSimple rejected",
			ctx:         context.Background(),
			runAtomic:   true,
			actionType:  "emailSimple",
			wantErr:     true,
			wantMention: "emailSimple",
		},
		{
			name:        "inherited atomic sendEmail rejected",
			ctx:         withAtomicFlowBoundary(context.Background()),
			runAtomic:   false,
			actionType:  flowSendEmailActionType,
			wantErr:     true,
			wantMention: flowSendEmailActionType,
		},
		{
			name:        "inherited atomic emailSimple rejected",
			ctx:         withAtomicFlowBoundary(context.Background()),
			runAtomic:   false,
			actionType:  "emailSimple",
			wantErr:     true,
			wantMention: "emailSimple",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			err := validateAtomicFlowForExecutionWithRegistry(tc.ctx, "", &models.FlowMetadata{
				RunAtomically: tc.runAtomic,
				ActionCalls: []models.FlowActionCall{{
					Name:       "SendEmail",
					ActionType: tc.actionType,
				}},
			}, newFlowActionCallRegistry(testRuntimeFlowActionCallRegistrations()...))
			if tc.wantErr {
				if err == nil {
					t.Fatal("validateAtomicFlowForExecution() error = nil, want error")
				}
				if !strings.Contains(err.Error(), tc.wantMention) {
					t.Fatalf("error = %v, want mention of %q", err, tc.wantMention)
				}
				return
			}
			if err != nil {
				t.Fatalf("validateAtomicFlowForExecution() error = %v, want nil", err)
			}
		})
	}
}

func TestValidateAtomicFlowForExecutionRejectsMetadataBackedActionCalls(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name        string
		ctx         context.Context
		runAtomic   bool
		actionName  string
		actionType  string
		wantErr     bool
		wantMention string
	}{
		{
			name:       "non-atomic metadata action allowed",
			ctx:        context.Background(),
			actionName: "WebhookAction",
			actionType: constants.ActionTypeExecuteAction,
		},
		{
			name:        "direct atomic action fallback rejected",
			ctx:         context.Background(),
			runAtomic:   true,
			actionName:  "WebhookAction",
			actionType:  constants.ActionTypeExecuteAction,
			wantErr:     true,
			wantMention: "metadata-backed action calls",
		},
		{
			name:        "direct atomic callWebhook metadata action rejected",
			ctx:         context.Background(),
			runAtomic:   true,
			actionName:  "WebhookAction",
			actionType:  constants.ActionTypeCallWebhook,
			wantErr:     true,
			wantMention: "WebhookAction",
		},
		{
			name:        "direct atomic composite metadata action rejected",
			ctx:         context.Background(),
			runAtomic:   true,
			actionName:  "CompositeAction",
			actionType:  constants.ActionTypeComposite,
			wantErr:     true,
			wantMention: "metadata-backed action calls",
		},
		{
			name:        "inherited atomic unknown metadata action rejected",
			ctx:         withAtomicFlowBoundary(context.Background()),
			actionName:  "MutableActionRow",
			actionType:  "customMetadataAction",
			wantErr:     true,
			wantMention: "MutableActionRow",
		},
		{
			name:       "direct atomic transfer connector remote identity internal action allowed",
			ctx:        context.Background(),
			runAtomic:  true,
			actionName: "TransferLeadConnectorRemoteIdentity",
			actionType: "transferConnectorRemoteIdentity",
		},
		{
			name:       "inherited atomic transfer connector remote identity internal action allowed",
			ctx:        withAtomicFlowBoundary(context.Background()),
			actionName: "TransferLeadConnectorRemoteIdentity",
			actionType: "transferConnectorRemoteIdentity",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			err := validateAtomicFlowForExecution(tc.ctx, "", &models.FlowMetadata{
				RunAtomically: tc.runAtomic,
				ActionCalls: []models.FlowActionCall{{
					Name:       "InvokeAction",
					ActionName: tc.actionName,
					ActionType: tc.actionType,
				}},
			})
			if tc.wantErr {
				if err == nil {
					t.Fatal("validateAtomicFlowForExecution() error = nil, want metadata-backed action rejection")
				}
				if !strings.Contains(err.Error(), tc.wantMention) {
					t.Fatalf("error = %v, want mention of %q", err, tc.wantMention)
				}
				return
			}
			if err != nil {
				t.Fatalf("validateAtomicFlowForExecution() error = %v, want nil", err)
			}
		})
	}
}

func TestExecuteRecordTriggerFlows_BeforeSaveAtomicFlowReusesOwnedTransaction(t *testing.T) {
	const (
		keyPrefix     = "00Q"
		recordID      = "00Q000000000001AAA"
		flowDefID     = "300000000000001AAA"
		flowVersionID = "301000000000001AAA"
	)

	persistence := &transactionalFanoutPersistence{}
	meta := models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Object:            "Lead",
			TriggerType:       "RecordBeforeSave",
			RecordTriggerType: "Update",
			Connector:         &models.FlowConnector{TargetReference: "AssignLeadCompany"},
		},
		Assignments: []models.FlowAssignment{{
			Name: "AssignLeadCompany",
			Items: []models.FlowAssignmentItem{{
				AssignToReference: "$Record.Company",
				Operator:          "Assign",
				Value:             stringValue("Mutated by atomic before-save"),
			}},
		}},
	}
	exec := fanoutExecutor(meta, persistence)
	version := &models.FlowVersion{
		Id:           flowVersionID,
		DefinitionId: flowDefID,
		Source:       "custom",
		Status:       models.FlowStatusActive,
		Metadata:     meta,
	}

	listener := &FlowTriggerListener{}
	listener.executor = exec
	listener.listTriggerSubscriptionsFn = func(ctx context.Context, kp string, et string, rtt string) ([]triggersub.Row, error) {
		return []triggersub.Row{{
			FlowDefinitionId:  flowDefID,
			FlowVersionId:     flowVersionID,
			KeyPrefix:         kp,
			EventType:         et,
			RecordTriggerType: rtt,
			IsActive:          true,
		}}, nil
	}
	listener.flowVersionsByIDFn = func(ctx context.Context, ids []string) (map[string]*models.FlowVersion, error) {
		return map[string]*models.FlowVersion{flowVersionID: version}, nil
	}
	listener.flowDefinitionByIDFn = func(ctx context.Context, id string) (*models.FlowDefinition, error) {
		return &models.FlowDefinition{
			Id:              id,
			DeveloperName:   "Atomic_BeforeSave",
			ActiveVersionId: flowVersionID,
		}, nil
	}
	listener.executeFlowVersionFn = func(ctx context.Context, version *models.FlowVersion, flowAPIName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
		return exec.executeFlowVersion(ctx, version, flowAPIName, inputs, user)
	}
	listener.recordTriggerKeyPrefixFn = func(ctx context.Context, payload eventFeature.RecordEventPayload, rid string) (string, error) {
		return keyPrefix, nil
	}

	payload := eventFeature.RecordEventPayload{
		ObjectAPIName: "Lead",
		Record: models.SObject{
			"Id":         recordID,
			"Company":    "Original Company",
			"attributes": map[string]interface{}{"type": "Lead"},
		},
	}

	err := persistence.RunInTransaction(context.Background(), func(txCtx context.Context) error {
		return listener.executeRecordTriggerFlows(txCtx, sharedEvents.RecordBeforeUpdate, payload)
	})
	if err != nil {
		t.Fatalf("executeRecordTriggerFlows() error = %v", err)
	}
	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 for owned before-save transaction reuse", persistence.runPessTxCalls)
	}
	if got := payload.Record["Company"]; got != "Mutated by atomic before-save" {
		t.Fatalf("payload.Record.Company = %v, want before-save assignment to land", got)
	}
}