package automation

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

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestExecuteFlowMetadataFailsInsteadOfCompletingPastExecutionBudget(t *testing.T) {
	executor := &FlowExecutor{
		formulaEngine:        formula.NewEngine(),
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(formula.NewEngine()),
		interviews:           NewFlowInterviewService(nil, nil),
	}
	flow := threeStepAssignmentFlow()
	interview := &models.FlowInterview{
		Id:          "budgeted-flow",
		ContextData: initializeFlowContext(flow, nil, "AutoLaunchedFlow"),
	}
	ctx, err := withFlowExecutionBudget(context.Background(), 2)
	if err != nil {
		t.Fatal(err)
	}

	_, _, err = executor.executeFlowMetadata(ctx, interview, flow, nil)
	if !errors.Is(err, ErrFlowExecutionBudgetExceeded) {
		t.Fatalf("executeFlowMetadata() error = %v, want budget exceeded", err)
	}
	if completed := interview.ContextData["_completedSteps"]; completed != float64(2) {
		t.Fatalf("completed steps = %v, want exactly two before rollback error", completed)
	}
}

func TestExecuteFlowMetadataHonorsCancellationBeforeFirstElement(t *testing.T) {
	executor := &FlowExecutor{
		formulaEngine:        formula.NewEngine(),
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(formula.NewEngine()),
		interviews:           NewFlowInterviewService(nil, nil),
	}
	flow := threeStepAssignmentFlow()
	interview := &models.FlowInterview{
		Id:          "cancelled-flow",
		ContextData: initializeFlowContext(flow, nil, "AutoLaunchedFlow"),
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	_, _, err := executor.executeFlowMetadata(ctx, interview, flow, nil)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("executeFlowMetadata() error = %v, want context canceled", err)
	}
	if completed, exists := interview.ContextData["_completedSteps"]; exists {
		t.Fatalf("cancelled Flow executed an element: completed steps = %v", completed)
	}
}

func TestPreparedFlowClosureKeepsSubflowOffLiveActiveVersionLookup(t *testing.T) {
	executor := NewFlowExecutor(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
	liveLookups := 0
	executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
		liveLookups++
		return nil, errors.New("live active-version lookup must not run")
	}
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: "root-definition", DefinitionRuntimeLoaded: true,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Source: "standard", DefinitionOwnerPack: "app-pack",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
			Subflows: []models.FlowSubflow{{
				Name: "CallChild", FlowName: "Child",
			}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition", DefinitionRuntimeLoaded: true,
		Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
		Source: "standard", DefinitionOwnerPack: "app-pack",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "SetValue"}},
			Assignments: []models.FlowAssignment{{
				Name: "SetValue",
			}},
		},
	}
	ctx, err := withPreparedFlowClosure(context.Background(), map[string]*models.FlowVersion{
		"Root": root, "Child": child,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx = WithTransientFlowInterviews(ctx)
	result, err := executor.ExecuteFlow(ctx, "Root", nil, &models.UserSession{ID: "user-1"})
	if err != nil {
		t.Fatalf("ExecuteFlow() error = %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("ExecuteFlow() result = %#v, want success", result)
	}
	if liveLookups != 0 {
		t.Fatalf("live active-version lookups = %d, want zero", liveLookups)
	}
}

func TestBoundedPreparedFlowClosureRejectsDirectCollectionBeforeAnyEffect(t *testing.T) {
	records := make([]models.SObject, MaxAtomicFilteredRecordOperationRows+1)
	for i := range records {
		records[i] = models.SObject{constants.FieldID: fmt.Sprintf("record-%03d", i)}
	}
	capture := &captureUpdate{}
	executor := minimalPersistenceWithCapture(capture)
	executor.interviews = NewFlowInterviewService(executor.persistence, nil)
	version := &models.FlowVersion{
		Id:                      "root-version",
		DefinitionId:            "root-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "UpdateRecords",
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "UpdateRecords", Object: "Widget", InputReference: "records",
			}},
		},
	}
	outcome, err := executor.ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Background(),
		"Root",
		map[string]*models.FlowVersion{"Root": version},
		map[string]interface{}{"records": records},
		&models.UserSession{ID: "system"},
		defaultFlowExecutionElementBudget,
	)
	if err != nil {
		t.Fatalf("bounded prepared execution error = %v", err)
	}
	if outcome == nil || outcome.Result == nil || outcome.Result.IsSuccess ||
		len(outcome.Result.Errors) == 0 {
		t.Fatalf("bounded prepared outcome = %#v, want fan-out rejection", outcome)
	}
	if len(capture.calls) != 0 {
		t.Fatalf("direct update applied %d effects before rejection", len(capture.calls))
	}
}

func TestResourceBoundedPreparedFlowClosureGovernsAuthorizedMultiVersionClosure(t *testing.T) {
	records := make([]models.SObject, MaxAtomicFilteredRecordOperationRows+1)
	for i := range records {
		records[i] = models.SObject{constants.FieldID: fmt.Sprintf("record-%03d", i)}
	}
	capture := &captureUpdate{}
	executor := minimalPersistenceWithCapture(capture)
	executor.interviews = NewFlowInterviewService(executor.persistence, nil)
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: "root-definition",
		DefinitionRuntimeLoaded: true, Status: models.FlowStatusActive,
		ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			RunAtomically: true, StartElementReference: "UpdateRecords",
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "UpdateRecords", Object: "Widget", InputReference: "records",
			}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition",
		DefinitionRuntimeLoaded: true, Status: models.FlowStatusActive,
		ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			RunAtomically: true, StartElementReference: "Done",
			Assignments: []models.FlowAssignment{{Name: "Done"}},
		},
	}
	outcome, err := executor.ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Background(),
		"Root",
		map[string]*models.FlowVersion{"Root": root, "Child": child},
		map[string]interface{}{"records": records},
		&models.UserSession{ID: "system"},
		defaultFlowExecutionElementBudget,
	)
	if err != nil {
		t.Fatalf("resource-bounded prepared execution error = %v", err)
	}
	if outcome == nil || outcome.Result == nil || outcome.Result.IsSuccess ||
		len(outcome.Result.Errors) == 0 {
		t.Fatalf("resource-bounded prepared outcome = %#v, want fan-out rejection", outcome)
	}
	if len(capture.calls) != 0 {
		t.Fatalf("direct update applied %d effects before rejection", len(capture.calls))
	}
}

func TestScheduledPreparedFlowInstallsDirectCollectionGovernor(t *testing.T) {
	records := make([]models.SObject, MaxAtomicFilteredRecordOperationRows+1)
	for i := range records {
		records[i] = models.SObject{constants.FieldID: fmt.Sprintf("record-%03d", i)}
	}
	capture := &captureUpdate{}
	executor := minimalPersistenceWithCapture(capture)
	executor.interviews = NewFlowInterviewService(executor.persistence, nil)
	version := &models.FlowVersion{
		Id: "scheduled-version", DefinitionId: "scheduled-definition",
		DefinitionRuntimeLoaded: true, Status: models.FlowStatusActive,
		ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			StartElementReference: "UpdateRecords",
			Variables: []models.FlowVariable{{
				Name: "records", DataType: "SObject", ObjectType: "Widget",
				IsCollection: true, Value: records,
			}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "UpdateRecords", Object: "Widget", InputReference: "records",
			}},
		},
	}
	scheduler := &SchedulerService{flowExecutor: executor}
	result, err := scheduler.executePreparedScheduledFlow(
		context.Background(),
		ScheduledFlowExecutionPlan{
			Definition: &models.FlowDefinition{
				Id: "scheduled-definition", DeveloperName: "Scheduled",
				ActiveVersionId: version.Id,
			},
			Root: version, Versions: map[string]*models.FlowVersion{"Scheduled": version},
			MaxElementExecutions: defaultFlowExecutionElementBudget,
		},
		nil,
		&models.UserSession{ID: "system"},
	)
	if err != nil {
		t.Fatalf("scheduled prepared execution error = %v", err)
	}
	if result == nil || result.IsSuccess || len(result.Errors) == 0 {
		t.Fatalf("scheduled prepared result = %#v, want fan-out rejection", result)
	}
	if len(capture.calls) != 0 {
		t.Fatalf("scheduled direct update applied %d effects before rejection", len(capture.calls))
	}
}

func TestOrdinaryScheduledPackFlowDoesNotRequireAsyncRootAuthority(t *testing.T) {
	executor := &FlowExecutor{
		formulaEngine:        formula.NewEngine(),
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(formula.NewEngine()),
		interviews:           NewFlowInterviewService(nil, nil),
	}
	version := &models.FlowVersion{
		Id: "scheduled-version", DefinitionId: "scheduled-definition",
		DefinitionRuntimeLoaded: true, DefinitionOwnerPack: "standard-pack",
		Source: flowSourceStandard, Status: models.FlowStatusActive,
		ProcessType: "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		},
	}
	result, err := (&SchedulerService{flowExecutor: executor}).executePreparedScheduledFlow(
		context.Background(),
		ScheduledFlowExecutionPlan{
			Definition: &models.FlowDefinition{
				Id: "scheduled-definition", DeveloperName: "Scheduled",
				ActiveVersionId: version.Id,
			},
			Root: version, Versions: map[string]*models.FlowVersion{"Scheduled": version},
			MaxElementExecutions: defaultFlowExecutionElementBudget,
		},
		nil,
		&models.UserSession{ID: "system"},
	)
	if err != nil || result == nil || !result.IsSuccess {
		t.Fatalf("ordinary scheduled pack flow result=%#v err=%v", result, err)
	}
}

func TestBoundedPreparedRecordLookupRejectsOverflowSentinel(t *testing.T) {
	records := make([]models.SObject, MaxAtomicFilteredRecordOperationRows+1)
	for i := range records {
		records[i] = models.SObject{constants.FieldID: fmt.Sprintf("record-%03d", i)}
	}
	query := &limitCapturingQuery{result: &models.QueryResult{Records: records}}
	executor := &FlowExecutor{query: query}
	ctx, err := withPreparedFlowDirectRecordOperationLimit(
		context.Background(),
		MaxAtomicFilteredRecordOperationRows,
	)
	if err != nil {
		t.Fatal(err)
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	lookup := &models.FlowRecordLookup{Name: "LoadRows", Object: "Widget"}
	if _, err := executor.executeRecordLookup(ctx, interview, lookup, nil); err == nil ||
		!strings.Contains(err.Error(), "exceeded prepared Flow governor limit") {
		t.Fatalf("bounded lookup overflow error = %v", err)
	}
	if query.capturedLimit != MaxAtomicFilteredRecordOperationRows+1 {
		t.Fatalf("bounded lookup query limit = %d, want overflow sentinel %d", query.capturedLimit, MaxAtomicFilteredRecordOperationRows+1)
	}
	if _, stored := interview.ContextData[lookup.Name]; stored {
		t.Fatal("overflowing lookup stored a truncated collection")
	}
}

func TestScheduledPreparedFlowRejectsCallerSuppliedCollection(t *testing.T) {
	scheduler := &SchedulerService{
		executePreparedScheduledFlowFn: func(
			context.Context,
			ScheduledFlowExecutionPlan,
			map[string]interface{},
			*models.UserSession,
		) (*models.FlowExecutionResult, error) {
			t.Fatal("scheduled executor ran with a caller-supplied collection")
			return nil, nil
		},
	}
	_, err := scheduler.executePreparedScheduledFlow(
		context.Background(),
		ScheduledFlowExecutionPlan{},
		map[string]interface{}{"records": []models.SObject{{constants.FieldID: "record-1"}}},
		&models.UserSession{ID: "system"},
	)
	if !errors.Is(err, ErrScheduledFlowNotEligible) {
		t.Fatalf("scheduled prepared execution error = %v, want ineligible collection input", err)
	}
}

func TestBoundedPreparedFlowClosureAcceptsExactSameOwnerNativeChildAuthority(t *testing.T) {
	executor := NewFlowExecutor(
		nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil,
	)
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: "root-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Source:                  "standard",
		DefinitionOwnerPack:     "companion-pack",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "CallChild",
			Subflows: []models.FlowSubflow{{
				Name: "CallChild", FlowName: "Child",
			}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Source:                  "standard",
		DefinitionOwnerPack:     "companion-pack",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		},
	}
	ctx := constants.WithConnectorInboundFlowExecution(
		context.Background(), "Root", "connector-pack",
	)
	ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, root.Id)
	outcome, err := executor.ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		"Root",
		map[string]*models.FlowVersion{"Root": root, "Child": child},
		nil,
		&models.UserSession{ID: "system"},
		10,
	)
	if err != nil {
		t.Fatalf("bounded prepared execution error = %v", err)
	}
	if outcome == nil || outcome.Result == nil || !outcome.Result.IsSuccess {
		t.Fatalf("bounded prepared outcome = %#v, want same-owner child success", outcome)
	}
}

func TestBoundedPreparedFlowClosureRejectsChildOutsideExactOneHopSameOwnerAuthority(t *testing.T) {
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: "root-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Source:                  "standard",
		DefinitionOwnerPack:     "companion-pack",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "CallChild",
			Subflows: []models.FlowSubflow{{
				Name: "CallChild", FlowName: "Child",
			}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Source:                  "standard",
		DefinitionOwnerPack:     "other-pack",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		},
	}
	ctx := constants.WithConnectorInboundCompanionFlowExecution(
		context.Background(), root.Id,
	)
	outcome, err := NewFlowExecutor(
		nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil,
	).ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		"Root",
		map[string]*models.FlowVersion{"Root": root, "Child": child},
		nil,
		&models.UserSession{ID: "system"},
		10,
	)
	if err == nil || !strings.Contains(err.Error(), "one-hop same-owner") {
		t.Fatalf(
			"bounded prepared outcome=%#v error=%v, want one-hop same-owner rejection",
			outcome,
			err,
		)
	}
}

func TestBoundedPreparedFlowClosureSharesExecutionBudgetWithDirectChild(t *testing.T) {
	executor := NewFlowExecutor(
		nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil,
	)
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: "root-definition",
		DefinitionRuntimeLoaded: true, Status: models.FlowStatusActive,
		ProcessType: "AutoLaunchedFlow", Source: "standard",
		DefinitionOwnerPack: "companion-pack",
		Metadata: models.FlowMetadata{
			RunAtomically: true, StartElementReference: "CallChild",
			Subflows: []models.FlowSubflow{{Name: "CallChild", FlowName: "Child"}},
		},
	}
	child := &models.FlowVersion{
		Id: "child-version", DefinitionId: "child-definition",
		DefinitionRuntimeLoaded: true, Status: models.FlowStatusActive,
		ProcessType: "AutoLaunchedFlow", Source: "standard",
		DefinitionOwnerPack: "companion-pack",
		Metadata: models.FlowMetadata{
			RunAtomically: true, StartElementReference: "Done",
			Assignments: []models.FlowAssignment{{Name: "Done"}},
		},
	}
	ctx := constants.WithConnectorInboundCompanionFlowExecution(
		context.Background(), root.Id,
	)
	outcome, err := executor.ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		"Root",
		map[string]*models.FlowVersion{"Root": root, "Child": child},
		nil,
		&models.UserSession{ID: "system"},
		1,
	)
	if !errors.Is(err, ErrFlowExecutionBudgetExceeded) {
		t.Fatalf(
			"bounded prepared outcome=%#v error=%v, want shared budget exhaustion",
			outcome,
			err,
		)
	}
}

func TestBoundedPreparedFlowClosureRejectsProtectedRootWithoutExactCompanionAuthority(
	t *testing.T,
) {
	executor := NewFlowExecutor(
		nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil,
	)
	root := &models.FlowVersion{
		Id: "root-version", DefinitionId: "root-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Source:                  "standard",
		DefinitionOwnerPack:     "companion-pack",
		Metadata: models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		},
	}
	ctx := constants.WithConnectorInboundFlowExecution(
		context.Background(), "Root", "connector-pack",
	)
	ctx = constants.WithConnectorInboundCompanionFlowExecution(
		ctx,
		"different-root-version",
	)
	outcome, err := executor.ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		"Root",
		map[string]*models.FlowVersion{"Root": root},
		nil,
		&models.UserSession{ID: "system"},
		10,
	)
	if err != nil {
		t.Fatalf("bounded prepared execution error = %v", err)
	}
	if outcome == nil || outcome.Result == nil || outcome.Result.IsSuccess ||
		len(outcome.Result.Errors) == 0 {
		t.Fatalf("bounded prepared outcome = %#v, want protected-root denial", outcome)
	}
}

func TestPreparedFlowClosureFailsClosedWhenSubflowIsAbsent(t *testing.T) {
	executor := NewFlowExecutor(nil, nil, nil, nil, nil, nil, nil)
	liveLookups := 0
	executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
		liveLookups++
		return &models.FlowVersion{Id: "live-child"}, nil
	}
	ctx, err := withPreparedFlowClosure(context.Background(), map[string]*models.FlowVersion{
		"Root": {Id: "root-version"},
	})
	if err != nil {
		t.Fatal(err)
	}
	_, err = executor.ExecuteFlow(ctx, "MissingChild", nil, &models.UserSession{ID: "user-1"})
	if !errors.Is(err, ErrPreparedFlowClosureMissing) {
		t.Fatalf("ExecuteFlow() error = %v, want prepared-closure failure", err)
	}
	if liveLookups != 0 {
		t.Fatalf("live active-version lookups = %d, want zero", liveLookups)
	}
}

func threeStepAssignmentFlow() *models.FlowMetadata {
	return &models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "First"},
		},
		Assignments: []models.FlowAssignment{
			{
				Name:      "First",
				Connector: &models.FlowConnector{TargetReference: "Second"},
			},
			{
				Name:      "Second",
				Connector: &models.FlowConnector{TargetReference: "Third"},
			},
			{Name: "Third"},
		},
	}
}