package automation

import (
	"errors"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/shared/pkg/models"
)

type scheduledFlowMetadataViewStub struct {
	definitions map[string]*models.FlowDefinition
	versions    map[string]*models.FlowVersion
	active      map[string]*models.FlowVersion
}

func (v scheduledFlowMetadataViewStub) FlowDefinition(id string) (*models.FlowDefinition, bool) {
	value, ok := v.definitions[id]
	return value, ok
}

func (v scheduledFlowMetadataViewStub) FlowVersion(id string) (*models.FlowVersion, bool) {
	value, ok := v.versions[id]
	return value, ok
}

func (v scheduledFlowMetadataViewStub) ActiveFlowVersion(name string) (*models.FlowVersion, bool) {
	value, ok := v.active[strings.ToLower(name)]
	return value, ok
}

func TestPrepareScheduledFlowExecutionPlanSelectsOneExactSafeClosure(t *testing.T) {
	rootDefinition := &models.FlowDefinition{
		Id: "root-definition", DeveloperName: "ScheduledRoot",
		ActiveVersionId: "root-version",
	}
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: rootDefinition.Id,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule:  &models.FlowSchedule{Frequency: "Daily"},
				Connector: &models.FlowConnector{TargetReference: "CallChild"},
			},
			Subflows: []models.FlowSubflow{{
				Name: "CallChild", FlowName: "ExactChild",
			}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition",
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Assign",
			Assignments:           []models.FlowAssignment{{Name: "Assign"}},
		},
	}
	view := scheduledFlowMetadataViewStub{
		definitions: map[string]*models.FlowDefinition{rootDefinition.Id: rootDefinition},
		versions:    map[string]*models.FlowVersion{root.Id: root},
		active:      map[string]*models.FlowVersion{"exactchild": child},
	}
	plan, err := PrepareScheduledFlowExecutionPlan(
		t.Context(),
		view,
		rootDefinition.Id,
		root.Id,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := plan.Validate(); err != nil {
		t.Fatal(err)
	}
	if len(plan.Versions) != 2 ||
		plan.Versions["scheduledroot"].Id != root.Id ||
		plan.Versions["exactchild"].Id != child.Id ||
		plan.MaxElementExecutions != scheduledFlowMaxElementExecutions {
		t.Fatalf("prepared plan = %#v", plan)
	}
}

func TestPrepareScheduledFlowExecutionPlanFailsClosedBeforeUnsafeEffects(t *testing.T) {
	for _, test := range []struct {
		name   string
		mutate func(*models.FlowMetadata)
	}{
		{
			name: "external email",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.ActionCalls = []models.FlowActionCall{{
					Name: "Send", ActionType: flowSendEmailActionType,
				}}
			},
		},
		{
			name: "wait",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.Waits = []models.FlowWait{{Name: "Wait"}}
			},
		},
		{
			name: "connector-sourced discovery",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.Start.ConnectorSource = &models.FlowConnectorSource{
					OwnerPack: "app", ProfileName: "directory",
				}
			},
		},
		{
			name: "unbounded record update input",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.RecordUpdates = []models.FlowRecordUpdate{{
					Name: "Update", InputReference: "requestCollection",
				}}
			},
		},
		{
			name: "caller supplied collection",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.Variables = []models.FlowVariable{{
					Name: "requestCollection", DataType: "SObject",
					ObjectType: "WorkItem", IsCollection: true, IsInput: true,
				}}
			},
		},
		{
			name: "duplicate collection declaration with conflicting type",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.Variables = []models.FlowVariable{
					{Name: "rows", DataType: "SObject", ObjectType: "WorkItem", IsCollection: true},
					{Name: "rows", DataType: "String", ObjectType: "WorkItem", IsCollection: true},
				}
			},
		},
		{
			name: "filtered update with whitespace input reference",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.RecordUpdates = []models.FlowRecordUpdate{{
					Name: "Update", Object: "WorkItem", InputReference: " ",
					Filters: []models.FlowRecordFilter{{Field: "State", Operator: "EqualTo"}},
				}}
			},
		},
		{
			name: "filtered delete with whitespace input reference",
			mutate: func(metadata *models.FlowMetadata) {
				metadata.RecordDeletes = []models.FlowRecordDelete{{
					Name: "Delete", Object: "WorkItem", InputReference: " ",
					Filters: []models.FlowRecordFilter{{Field: "State", Operator: "EqualTo"}},
				}}
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			definition := &models.FlowDefinition{
				Id: "definition", DeveloperName: "Scheduled",
				ActiveVersionId: "version",
			}
			version := &models.FlowVersion{
				Id: "version", DefinitionId: definition.Id,
				Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{Frequency: "Daily"},
					},
					StartElementReference: "Assign",
					Assignments:           []models.FlowAssignment{{Name: "Assign"}},
				},
			}
			test.mutate(&version.Metadata)
			view := scheduledFlowMetadataViewStub{
				definitions: map[string]*models.FlowDefinition{definition.Id: definition},
				versions:    map[string]*models.FlowVersion{version.Id: version},
			}
			if _, err := PrepareScheduledFlowExecutionPlan(
				t.Context(),
				view,
				definition.Id,
				version.Id,
			); !errors.Is(err, ErrScheduledFlowNotEligible) {
				t.Fatalf("PrepareScheduledFlowExecutionPlan() error = %v", err)
			}
		})
	}
}

func TestPrepareScheduledFlowExecutionPlanAllowsBudgetedDatabaseCollectionLoop(t *testing.T) {
	definition := &models.FlowDefinition{
		Id: "definition", DeveloperName: "Scheduled",
		ActiveVersionId: "version",
	}
	version := &models.FlowVersion{
		Id: "version", DefinitionId: definition.Id,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Schedule:  &models.FlowSchedule{Frequency: "Daily"},
				Connector: &models.FlowConnector{TargetReference: "LoadRows"},
			},
			RecordLookups: []models.FlowRecordLookup{{
				Name: "LoadRows", Object: "WorkItem", GetFirstRecordOnly: false,
				Connector: &models.FlowConnector{TargetReference: "EachRow"},
			}},
			Loops: []models.FlowLoop{{
				Name: "EachRow", CollectionReference: "LoadRows",
				NextValueConnector:    &models.FlowConnector{TargetReference: "CollectRow"},
				NoMoreValuesConnector: &models.FlowConnector{TargetReference: "UpdateRecords"},
			}},
			Variables: []models.FlowVariable{{
				Name: "RowsToUpdate", DataType: " SObject ", ObjectType: "WorkItem",
				IsCollection: true,
			}},
			Assignments: []models.FlowAssignment{{
				Name: "CollectRow",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "RowsToUpdate", Operator: "Add",
					Value: elementReferenceValue("EachRow"),
				}},
				Connector: &models.FlowConnector{TargetReference: "EachRow"},
			}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "UpdateRecords", InputReference: "RowsToUpdate",
				Connector: &models.FlowConnector{TargetReference: "UpdateFiltered"},
			}, {
				Name: "UpdateFiltered", Object: "WorkItem",
				Filters: []models.FlowRecordFilter{{Field: "State", Operator: "EqualTo"}},
			}},
		},
	}
	view := scheduledFlowMetadataViewStub{
		definitions: map[string]*models.FlowDefinition{definition.Id: definition},
		versions:    map[string]*models.FlowVersion{version.Id: version},
	}
	plan, err := PrepareScheduledFlowExecutionPlan(
		t.Context(),
		view,
		definition.Id,
		version.Id,
	)
	if err != nil {
		t.Fatal(err)
	}
	if plan.MaxElementExecutions != scheduledFlowMaxElementExecutions {
		t.Fatalf("MaxElementExecutions = %d", plan.MaxElementExecutions)
	}
}

func TestPrepareTimedFlowExecutionPlanAllowsBoundedCollectionSubflow(t *testing.T) {
	definition := &models.FlowDefinition{
		Id: "definition", DeveloperName: "RecordAsyncRoot", ActiveVersionId: "root-version",
	}
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: definition.Id,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Object: "WorkItem", TriggerType: "RecordAfterSave", RecordTriggerType: "Create",
				ScheduledPaths: []models.FlowScheduledPath{{
					PathType:  "AsyncAfterCommit",
					Connector: &models.FlowConnector{TargetReference: "CallCollectionWorker"},
				}},
			},
			Subflows: []models.FlowSubflow{{
				Name: "CallCollectionWorker", FlowName: "CollectionWorker",
			}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition",
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "LoadRows"}},
			Variables: []models.FlowVariable{{
				Name: "RowsToUpdate", DataType: "SObject", ObjectType: "WorkItem", IsCollection: true,
			}},
			RecordLookups: []models.FlowRecordLookup{{
				Name: "LoadRows", Object: "WorkItem", GetFirstRecordOnly: false,
				Connector: &models.FlowConnector{TargetReference: "EachRow"},
			}},
			Loops: []models.FlowLoop{{
				Name: "EachRow", CollectionReference: "LoadRows",
				NextValueConnector:    &models.FlowConnector{TargetReference: "CollectRow"},
				NoMoreValuesConnector: &models.FlowConnector{TargetReference: "UpdateRows"},
			}},
			Assignments: []models.FlowAssignment{{
				Name: "CollectRow",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "RowsToUpdate", Operator: "Add",
					Value: elementReferenceValue("EachRow"),
				}},
				Connector: &models.FlowConnector{TargetReference: "EachRow"},
			}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "UpdateRows", InputReference: "RowsToUpdate",
				Connector: &models.FlowConnector{TargetReference: "UpdateByFilter"},
			}, {
				Name: "UpdateByFilter", Object: "WorkItem",
				Filters: []models.FlowRecordFilter{{Field: "Id", Operator: "EqualTo"}},
			}},
		},
	}
	view := scheduledFlowMetadataViewStub{
		definitions: map[string]*models.FlowDefinition{definition.Id: definition},
		versions:    map[string]*models.FlowVersion{root.Id: root},
		active:      map[string]*models.FlowVersion{"collectionworker": child},
	}
	plan, err := PrepareScheduledFlowExecutionPlanForEntry(
		t.Context(), view, definition.Id, root.Id, "CallCollectionWorker",
	)
	if err != nil {
		t.Fatal(err)
	}
	if !plan.AsyncAfterCommit || len(plan.Versions) != 2 ||
		plan.Versions["collectionworker"].Id != child.Id {
		t.Fatalf("prepared timed collection plan = %#v", plan)
	}
}

func TestPrepareScheduledFlowExecutionPlanForEntrySelectsOnlyExactTimedPath(t *testing.T) {
	definition := &models.FlowDefinition{
		Id: "definition", DeveloperName: "Timed",
		ActiveVersionId: "version",
	}
	version := &models.FlowVersion{
		Id: "version", DefinitionId: definition.Id,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Object:      "WorkItem",
				TriggerType: "RecordAfterSave",
				Connector:   &models.FlowConnector{TargetReference: "RunNow"},
				Filters:     []models.FlowRecordFilter{{Field: "State", Operator: "EqualTo"}},
				ScheduledPaths: []models.FlowScheduledPath{{
					Name: "OneMinuteLater",
					Connector: &models.FlowConnector{
						TargetReference: "RunLater",
					},
					TimeSource: "RecordTriggerEvent", OffsetNumber: 1, OffsetUnit: "Minutes",
				}},
			},
			Assignments: []models.FlowAssignment{
				{Name: "RunNow"},
				{Name: "RunLater"},
			},
		},
	}
	view := scheduledFlowMetadataViewStub{
		definitions: map[string]*models.FlowDefinition{definition.Id: definition},
		versions:    map[string]*models.FlowVersion{version.Id: version},
	}
	plan, err := PrepareScheduledFlowExecutionPlanForEntry(
		t.Context(),
		view,
		definition.Id,
		version.Id,
		"RunLater",
	)
	if err != nil {
		t.Fatal(err)
	}
	if plan.EntryElement != "RunLater" ||
		plan.Root.Metadata.Start.Connector == nil ||
		plan.Root.Metadata.Start.Connector.TargetReference != "RunLater" ||
		plan.Root.Metadata.Start.TriggerType != "" ||
		len(plan.Root.Metadata.Start.Filters) != 0 ||
		len(plan.Root.Metadata.Start.ScheduledPaths) != 0 {
		t.Fatalf("timed execution plan = %#v", plan)
	}
	if version.Metadata.Start.Connector.TargetReference != "RunNow" ||
		len(version.Metadata.Start.ScheduledPaths) != 1 {
		t.Fatal("preparing the timed path mutated immutable Catalog metadata")
	}
	if _, err := PrepareScheduledFlowExecutionPlanForEntry(
		t.Context(),
		view,
		definition.Id,
		version.Id,
		"Missing",
	); !errors.Is(err, ErrScheduledFlowNotEligible) {
		t.Fatalf("missing timed entry error = %v", err)
	}
}

func TestPrepareScheduledFlowExecutionPlanForEntryAcceptsOnAllChangesAlias(t *testing.T) {
	definition := &models.FlowDefinition{
		Id: "definition", DeveloperName: "Timed",
		ActiveVersionId: "version",
	}
	version := &models.FlowVersion{
		Id: "version", DefinitionId: definition.Id,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Object: "WorkItem", TriggerType: "onAllChanges",
				ScheduledPaths: []models.FlowScheduledPath{{
					PathType:  "AsyncAfterCommit",
					Connector: &models.FlowConnector{TargetReference: "RunAsync"},
				}},
			},
			Assignments: []models.FlowAssignment{{Name: "RunAsync"}},
		},
	}
	view := scheduledFlowMetadataViewStub{
		definitions: map[string]*models.FlowDefinition{definition.Id: definition},
		versions:    map[string]*models.FlowVersion{version.Id: version},
	}
	plan, err := PrepareScheduledFlowExecutionPlanForEntry(
		t.Context(), view, definition.Id, version.Id, "RunAsync",
	)
	if err != nil {
		t.Fatalf("onAllChanges timed execution plan rejected: %v", err)
	}
	if plan.EntryElement != "RunAsync" ||
		plan.Root.Metadata.Start == nil ||
		plan.Root.Metadata.Start.Connector == nil ||
		plan.Root.Metadata.Start.Connector.TargetReference != "RunAsync" {
		t.Fatalf("onAllChanges timed execution plan = %#v", plan)
	}
}

func TestPrepareScheduledFlowExecutionPlanForEntryAllowsOnlyAtomicAsyncAfterCommitIdentityActions(t *testing.T) {
	definition := &models.FlowDefinition{
		Id: "definition", DeveloperName: "IdentityHandoff", ActiveVersionId: "version",
	}
	newVersion := func(runAtomically bool, actionType string) *models.FlowVersion {
		return &models.FlowVersion{
			Id: "version", DefinitionId: definition.Id, Status: models.FlowStatusActive,
			ProcessType: "AutoLaunchedFlow",
			Metadata: models.FlowMetadata{
				RunAtomically: runAtomically,
				Start: &models.FlowStart{Object: "Contact", TriggerType: "RecordAfterSave",
					ScheduledPaths: []models.FlowScheduledPath{{PathType: "AsyncAfterCommit", Connector: &models.FlowConnector{TargetReference: "Resolve"}}},
				},
				ActionCalls: []models.FlowActionCall{
					{Name: "Resolve", ActionType: flowactions.ActionTypeResolveConnectorRemoteIdentity, Connector: &models.FlowConnector{TargetReference: "Transfer"}},
					{Name: "Transfer", ActionType: actionType},
				},
			},
		}
	}
	for _, test := range []struct {
		name          string
		runAtomically bool
		actionType    string
		wantErr       bool
	}{
		{name: "atomic identity actions", runAtomically: true, actionType: flowactions.ActionTypeTransferConnectorRemoteIdentity},
		{name: "non-atomic identity actions", runAtomically: false, actionType: flowactions.ActionTypeTransferConnectorRemoteIdentity, wantErr: true},
		{name: "push remains owned by outbound worker", runAtomically: true, actionType: "PushToConnector", wantErr: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			version := newVersion(test.runAtomically, test.actionType)
			view := scheduledFlowMetadataViewStub{
				definitions: map[string]*models.FlowDefinition{definition.Id: definition},
				versions:    map[string]*models.FlowVersion{version.Id: version},
			}
			plan, err := PrepareScheduledFlowExecutionPlanForEntry(t.Context(), view, definition.Id, version.Id, "Resolve")
			if test.wantErr {
				if !errors.Is(err, ErrScheduledFlowNotEligible) {
					t.Fatalf("PrepareScheduledFlowExecutionPlanForEntry error = %v, want ErrScheduledFlowNotEligible", err)
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if !plan.AsyncAfterCommit {
				t.Fatalf("plan = %#v, want strict AsyncAfterCommit marker", plan)
			}
		})
	}
}