package runtimeworker

import (
	"errors"
	"strings"
	"testing"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

func TestRuntimeFlowApplicationCommandV3ExecutesExactSuppressedFlowAsSystemActor(t *testing.T) {
	_, _, flowView := workerScheduledFlowView(t)
	engine := &runtimeFlowCommandEngineProbe{
		outcome: &contractmodels.FlowExecutionTypedOutcome{
			ExecutionID: "flow-execution-1",
			Result: &contractmodels.FlowExecutionResult{
				ActionName: "ExactChild",
				IsSuccess:  true,
				OutputValues: map[string]interface{}{
					"z": "last",
					"a": "first",
				},
			},
		},
	}
	users := &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}}
	executor := &runtimeFlowCommandApplicationExecutor{flows: engine, users: users}
	payload := runtimeFlowApplicationCommandV3Payload(
		t,
		flowcommandsemanticv3.SuppressExternalEffects,
		[]byte(`{"count":2,"name":"input"}`),
	)
	completion, err := executor.ExecutePlatformApplicationCommand(
		metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
		PlatformCommandExecution{
			CommandKind:          flowcommandsemanticv3.CommandKind,
			PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
			Payload:              payload,
			CallerWorkloadID:     "integration-a",
			DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{
				allowed: true, payloadVersion: flowcommandsemanticv3.PayloadFormatVersion,
			},
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if completion.Outcome != applicationresultsemantic.OutcomeApplied ||
		completion.FailureCode != applicationresultsemantic.FailureNone ||
		string(completion.Body) != `{"canonicalOutputs":{"a":"first","z":"last"},"executionId":"flow-execution-1"}` {
		t.Fatalf("completion = %#v", completion)
	}
	if strings.Contains(string(completion.Body), "connector-execution-1") {
		t.Fatalf("Flow result body copied Connector ExecutionID: %s", completion.Body)
	}
	if engine.calls != 1 || engine.name != "ExactChild" || engine.version != "version-child" ||
		engine.user == nil || engine.user.ID != constants.SystemUserID || !engine.system ||
		engine.connectorNamespace != "connector-a" || !engine.inboundRootAuthorized ||
		!engine.inboundVersionAuthorized || users.calls != 1 {
		t.Fatalf("execution = %#v users=%d", engine, users.calls)
	}
}

func TestRuntimeFlowApplicationCommandV3RefusesAllowedExternalEffectsBeforeFlowExecution(t *testing.T) {
	engine := &runtimeFlowCommandEngineProbe{}
	users := &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}}
	executor := &runtimeFlowCommandApplicationExecutor{flows: engine, users: users}
	completion, err := executor.ExecutePlatformApplicationCommand(t.Context(), PlatformCommandExecution{
		CommandKind:          flowcommandsemanticv3.CommandKind,
		PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
		Payload: runtimeFlowApplicationCommandV3Payload(
			t,
			flowcommandsemanticv3.AllowExternalEffects,
			[]byte(`{}`),
		),
	})
	var executionErr *PlatformCommandExecutionError
	if !errors.As(err, &executionErr) || !executionErr.Permanent ||
		!runtimeFlowApplicationCompletionIsZero(completion) ||
		engine.calls != 0 || users.calls != 0 {
		t.Fatalf("completion=%#v err=%v engine=%d users=%d", completion, err, engine.calls, users.calls)
	}
}

func TestRuntimeFlowApplicationCommandV3RejectsMalformedPayloadBeforeFlowExecution(t *testing.T) {
	engine := &runtimeFlowCommandEngineProbe{}
	users := &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}}
	executor := &runtimeFlowCommandApplicationExecutor{flows: engine, users: users}
	completion, err := executor.ExecutePlatformApplicationCommand(t.Context(), PlatformCommandExecution{
		CommandKind:          flowcommandsemanticv3.CommandKind,
		PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
		Payload:              []byte(`{"canonicalInputs":{},"unknown":true}`),
	})
	var executionErr *PlatformCommandExecutionError
	if !errors.As(err, &executionErr) || !executionErr.Permanent ||
		!runtimeFlowApplicationCompletionIsZero(completion) || engine.calls != 0 || users.calls != 0 {
		t.Fatalf("completion=%#v err=%v engine=%d users=%d", completion, err, engine.calls, users.calls)
	}
}

func TestRuntimeFlowApplicationCommandV3MapsOnlyProvedTerminalOutcomes(t *testing.T) {
	for _, test := range []struct {
		name        string
		outcome     *contractmodels.FlowExecutionTypedOutcome
		engineErr   error
		wantOutcome applicationresultsemantic.Outcome
		wantFailure applicationresultsemantic.FailureCode
		wantError   bool
		wantRetry   bool
		wantBody    string
	}{
		{
			name: "entry skip output remains applied rather than inferred noop",
			outcome: runtimeFlowApplicationOutcome(true, false, false, nil, map[string]interface{}{
				"_skipped": true,
			}),
			wantOutcome: applicationresultsemantic.OutcomeApplied,
			wantBody:    `{"canonicalOutputs":{"_skipped":true},"executionId":"flow-execution-1"}`,
		},
		{
			name: "custom error is application rejection",
			outcome: runtimeFlowApplicationOutcome(false, true, false, &contractmodels.FlowExecutionCause{
				Code: "FLOW_CUSTOM_ERROR", Category: contractmodels.FlowExecutionCauseCategoryUnknown,
			}, nil),
			engineErr:   errors.New("Flow surfaced its custom error"),
			wantOutcome: applicationresultsemantic.OutcomeRejected,
			wantFailure: applicationresultsemantic.FailureApplicationRejected,
			wantBody:    `{"causeCategory":"unknown","causeCode":"FLOW_CUSTOM_ERROR","executionId":"flow-execution-1"}`,
		},
		{
			name: "validation is application rejection",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "VALIDATION_RULE", Category: contractmodels.FlowExecutionCauseCategoryValidation,
			}, nil),
			engineErr:   errors.New("Flow surfaced its validation error"),
			wantOutcome: applicationresultsemantic.OutcomeRejected,
			wantFailure: applicationresultsemantic.FailureApplicationRejected,
			wantBody:    `{"causeCategory":"validation","causeCode":"VALIDATION_RULE","executionId":"flow-execution-1"}`,
		},
		{
			name: "authorization is application rejection",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "AUTHORIZATION_DENIED", Category: contractmodels.FlowExecutionCauseCategoryAuthorization,
			}, nil),
			wantOutcome: applicationresultsemantic.OutcomeRejected,
			wantFailure: applicationresultsemantic.FailureApplicationRejected,
			wantBody:    `{"causeCategory":"authorization","causeCode":"AUTHORIZATION_DENIED","executionId":"flow-execution-1"}`,
		},
		{
			name: "not found is application rejection",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "NOT_FOUND", Category: contractmodels.FlowExecutionCauseCategoryNotFound,
			}, nil),
			wantOutcome: applicationresultsemantic.OutcomeRejected,
			wantFailure: applicationresultsemantic.FailureApplicationRejected,
			wantBody:    `{"causeCategory":"not_found","causeCode":"NOT_FOUND","executionId":"flow-execution-1"}`,
		},
		{
			name: "runtime configuration failure is failed",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "INVALID_FLOW", Category: contractmodels.FlowExecutionCauseCategoryConfiguration,
			}, nil),
			engineErr:   errors.New("Flow surfaced its runtime failure"),
			wantOutcome: applicationresultsemantic.OutcomeFailed,
			wantFailure: applicationresultsemantic.FailureRuntimeExecutionFailed,
			wantBody:    `{"causeCategory":"configuration","causeCode":"INVALID_FLOW","executionId":"flow-execution-1"}`,
		},
		{
			name: "non-retriable persistence failure is failed",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "DB_ERROR", Category: contractmodels.FlowExecutionCauseCategoryPersistence,
			}, nil),
			wantOutcome: applicationresultsemantic.OutcomeFailed,
			wantFailure: applicationresultsemantic.FailureRuntimeExecutionFailed,
			wantBody:    `{"causeCategory":"persistence","causeCode":"DB_ERROR","executionId":"flow-execution-1"}`,
		},
		{
			name: "non-retriable external failure is failed",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "EXTERNAL_ERROR", Category: contractmodels.FlowExecutionCauseCategoryExternal,
			}, nil),
			wantOutcome: applicationresultsemantic.OutcomeFailed,
			wantFailure: applicationresultsemantic.FailureRuntimeExecutionFailed,
			wantBody:    `{"causeCategory":"external","causeCode":"EXTERNAL_ERROR","executionId":"flow-execution-1"}`,
		},
		{
			name: "non-retriable internal failure is failed",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "INTERNAL_ERROR", Category: contractmodels.FlowExecutionCauseCategoryInternal,
			}, nil),
			wantOutcome: applicationresultsemantic.OutcomeFailed,
			wantFailure: applicationresultsemantic.FailureRuntimeExecutionFailed,
			wantBody:    `{"causeCategory":"internal","causeCode":"INTERNAL_ERROR","executionId":"flow-execution-1"}`,
		},
		{
			name: "classified unknown failure is failed",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "FLOW_ERROR", Category: contractmodels.FlowExecutionCauseCategoryUnknown,
			}, nil),
			wantOutcome: applicationresultsemantic.OutcomeFailed,
			wantFailure: applicationresultsemantic.FailureRuntimeExecutionFailed,
			wantBody:    `{"causeCategory":"unknown","causeCode":"FLOW_ERROR","executionId":"flow-execution-1"}`,
		},
		{
			name:      "ordinary engine error has no terminal result",
			engineErr: errors.New("engine failed before terminal evidence"),
			wantError: true,
		},
		{
			name: "malformed successful result has no terminal result",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				ExecutionID: "flow-execution-1",
				Result: &contractmodels.FlowExecutionResult{
					IsSuccess: true, IsCustomError: true,
				},
			},
			wantError: true,
		},
		{
			name: "typed retriable failure has no terminal result",
			outcome: runtimeFlowApplicationOutcome(false, false, true, &contractmodels.FlowExecutionCause{
				Code: "DB_WRITE_CONFLICT", Category: contractmodels.FlowExecutionCauseCategoryPersistence,
			}, nil),
			engineErr: errors.New("Flow surfaced its retriable failure"),
			wantError: true,
			wantRetry: true,
		},
		{
			name: "typed retriable cause has no terminal result",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "REMOTE_RETRY", Category: contractmodels.FlowExecutionCauseCategoryExternal,
				Retriable: true,
			}, nil),
			wantError: true,
			wantRetry: true,
		},
		{
			name: "concurrency is retryable even with a false flag",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "DB_DEADLOCK", Category: contractmodels.FlowExecutionCauseCategoryConcurrency,
			}, nil),
			wantError: true,
			wantRetry: true,
		},
		{
			name: "unknown category string has no terminal result",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: "NEW_FAILURE", Category: contractmodels.FlowExecutionCauseCategory("new_category"),
			}, nil),
			wantError: true,
		},
		{
			name: "noncanonical cause code has no terminal result",
			outcome: runtimeFlowApplicationOutcome(false, false, false, &contractmodels.FlowExecutionCause{
				Code: " FLOW_ERROR ", Category: contractmodels.FlowExecutionCauseCategoryUnknown,
			}, nil),
			wantError: true,
		},
		{
			name: "success-shaped result with engine error has no terminal result",
			outcome: runtimeFlowApplicationOutcome(true, false, false, &contractmodels.FlowExecutionCause{
				Code: "INTERNAL_ERROR", Category: contractmodels.FlowExecutionCauseCategoryInternal,
			}, nil),
			engineErr: errors.New("engine contradicted its success-shaped result"),
			wantError: true,
		},
		{
			name:      "missing outcome has no terminal result",
			wantError: true,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			_, _, flowView := workerScheduledFlowView(t)
			engine := &runtimeFlowCommandEngineProbe{outcome: test.outcome, err: test.engineErr}
			executor := &runtimeFlowCommandApplicationExecutor{
				flows: engine,
				users: &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}},
			}
			completion, err := executor.ExecutePlatformApplicationCommand(
				metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
				PlatformCommandExecution{
					CommandKind:          flowcommandsemanticv3.CommandKind,
					PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
					Payload: runtimeFlowApplicationCommandV3Payload(
						t,
						flowcommandsemanticv3.SuppressExternalEffects,
						[]byte(`{}`),
					),
					CallerWorkloadID: "integration-a",
					DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{
						allowed: true, payloadVersion: flowcommandsemanticv3.PayloadFormatVersion,
					},
				},
			)
			if (err != nil) != test.wantError {
				t.Fatalf("completion=%#v err=%v", completion, err)
			}
			if test.wantError {
				if !runtimeFlowApplicationCompletionIsZero(completion) {
					t.Fatalf("ordinary error minted completion %#v", completion)
				}
				if test.wantRetry {
					var executionErr *PlatformCommandExecutionError
					if !errors.As(err, &executionErr) || executionErr.Permanent {
						t.Fatalf("err=%v, want non-permanent execution error", err)
					}
				}
				return
			}
			if completion.Outcome != test.wantOutcome || completion.FailureCode != test.wantFailure ||
				string(completion.Body) != test.wantBody || strings.Contains(string(completion.Body), "Message") ||
				strings.Contains(string(completion.Body), "do not persist") {
				t.Fatalf("completion=%#v", completion)
			}
		})
	}
}

func TestRuntimeFlowApplicationCommandV3LeavesDependencyAndRecoveryNonterminal(t *testing.T) {
	dependency := &contractmodels.FlowExecutionDependencyRequest{
		Kind:          contractmodels.FlowDependencyKindIdentityConvergence,
		ObjectAPIName: "Lead", RecordID: "lead-1", ConnectorOwnerKey: "connector-a",
		DestinationSlotID: "contact",
	}
	recovery := &automation.ConnectorInboundRemoteIdentityRecoveryRequestedError{
		Selector: automation.ConnectorInboundRemoteIdentityRecoverySelector{
			ConnectorOwnerKey: "connector-a", ConnectorNamespace: "connector-a",
			DestinationSlotID: "contact", RemoteID: "remote-1",
		},
	}
	for _, test := range []struct {
		name      string
		outcome   *contractmodels.FlowExecutionTypedOutcome
		engineErr error
		recovery  bool
	}{
		{
			name: "unexpected dependency",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				ExecutionID: "flow-execution-1",
				Result:      &contractmodels.FlowExecutionResult{IsSuccess: true},
				Dependency:  dependency,
			},
		},
		{
			name: "recovery request",
			outcome: runtimeFlowApplicationOutcome(false, false, true, &contractmodels.FlowExecutionCause{
				Code: "RECOVERY", Category: contractmodels.FlowExecutionCauseCategoryExternal,
			}, nil),
			engineErr: recovery,
			recovery:  true,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			_, _, flowView := workerScheduledFlowView(t)
			executor := &runtimeFlowCommandApplicationExecutor{
				flows: &runtimeFlowCommandEngineProbe{outcome: test.outcome, err: test.engineErr},
				users: &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}},
			}
			completion, err := executor.ExecutePlatformApplicationCommand(
				metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
				PlatformCommandExecution{
					CommandKind:          flowcommandsemanticv3.CommandKind,
					PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
					Payload: runtimeFlowApplicationCommandV3Payload(
						t, flowcommandsemanticv3.SuppressExternalEffects, []byte(`{}`),
					),
					CallerWorkloadID: "integration-a",
					DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{
						allowed: true, payloadVersion: flowcommandsemanticv3.PayloadFormatVersion,
					},
				},
			)
			if err == nil || !runtimeFlowApplicationCompletionIsZero(completion) {
				t.Fatalf("completion=%#v err=%v", completion, err)
			}
			if test.recovery {
				var recoveryErr *PlatformCommandRecoveryError
				if !errors.As(err, &recoveryErr) || recoveryErr.Request.Selector != recovery.Selector {
					t.Fatalf("recovery err=%#v", err)
				}
			}
		})
	}
}

func runtimeFlowApplicationOutcome(
	success bool,
	custom bool,
	retriable bool,
	cause *contractmodels.FlowExecutionCause,
	outputs map[string]interface{},
) *contractmodels.FlowExecutionTypedOutcome {
	var executionErrors []contractmodels.FlowExecutionError
	if !success {
		executionErrors = []contractmodels.FlowExecutionError{{
			StatusCode: "STABLE_CODE", Message: "do not persist this Message",
		}}
	}
	return &contractmodels.FlowExecutionTypedOutcome{
		ExecutionID: "flow-execution-1",
		Result: &contractmodels.FlowExecutionResult{
			ActionName: "ExactChild", IsSuccess: success, IsCustomError: custom,
			IsRetriable: retriable, OutputValues: outputs,
			Errors: executionErrors,
		},
		Cause: cause,
	}
}

func runtimeFlowApplicationCommandV3Payload(
	t *testing.T,
	mode flowcommandsemanticv3.ExternalEffectsMode,
	inputs []byte,
) []byte {
	t.Helper()
	inputDigest, err := flowcommandsemanticv3.ComputeInputDigest(inputs, mode)
	if err != nil {
		t.Fatal(err)
	}
	payload, err := flowcommandsemanticv3.Encode(flowcommandsemanticv3.Payload{
		ExecutionRef: applicationresultsemantic.ExecutionRef{
			ExecutionID: "connector-execution-1", PackGenerationID: "pack-generation-1",
			ServingEpochDigest: "sha256:" + strings.Repeat("a", 64),
			DefinitionRef:      "ExactChild", DefinitionVersion: "version-child",
			InputDigest: inputDigest,
		},
		CanonicalInputs:     inputs,
		ExternalEffectsMode: mode,
	})
	if err != nil {
		t.Fatal(err)
	}
	return payload
}

func runtimeFlowApplicationCompletionIsZero(
	completion persistence.RuntimePlatformCommandApplicationCompletion,
) bool {
	return completion.Outcome == "" && completion.FailureCode == "" && len(completion.Body) == 0
}