package runtimeworker

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"

	"github.com/nexuscrm/automation"
	flowexecution "github.com/nexuscrm/backend/internal/runtimeflowexecution"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	flowcommandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

type runtimeFlowCommandEngine interface {
	ValidateAtomicFlowForExecution(
		context.Context,
		string,
		*models.FlowMetadata,
	) error
	ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Context,
		string,
		map[string]*models.FlowVersion,
		map[string]interface{},
		*models.UserSession,
		uint64,
	) (*contractmodels.FlowExecutionTypedOutcome, error)
}

type runtimeFlowCommandSystemUserLoader interface {
	LoadSystemUserSession(context.Context) (*models.UserSession, error)
}

type runtimeFlowCommandExecutor struct {
	flows runtimeFlowCommandEngine
	users runtimeFlowCommandSystemUserLoader
}

type runtimeFlowCommandDependencyExecutor struct {
	flows runtimeFlowCommandEngine
	users runtimeFlowCommandSystemUserLoader
}

type runtimeFlowCommandPayload struct {
	definitionRef     string
	definitionVersion string
	canonicalInputs   []byte
	dependencySource  *flowcommandsemanticv2.DependencySourceScope
}

type runtimeFlowCommandResult struct {
	ExecutionID      string          `json:"executionId"`
	CanonicalOutputs json.RawMessage `json:"canonicalOutputs"`
}

func (e *runtimeFlowCommandExecutor) ExecutePlatformCommand(
	ctx context.Context,
	command PlatformCommandExecution,
) ([]byte, error) {
	if command.PayloadFormatVersion != RuntimeFlowCommandPayloadFormatVersion {
		return nil, permanentPlatformCommandError(
			errors.New("runtime Flow command executor is not configured for the command"),
		)
	}
	payload, inputs, err := decodeRuntimeFlowCommandPayload(command.Payload)
	if err != nil {
		return nil, permanentPlatformCommandError(err)
	}
	return executeRuntimeFlowCommand(
		ctx,
		command,
		e.flows,
		e.users,
		runtimeFlowCommandPayload{
			definitionRef:     payload.DefinitionRef,
			definitionVersion: payload.DefinitionVersion,
			canonicalInputs:   payload.CanonicalInputs,
		},
		inputs,
		false,
	)
}

func (e *runtimeFlowCommandDependencyExecutor) ExecutePlatformCommand(
	ctx context.Context,
	command PlatformCommandExecution,
) ([]byte, error) {
	if command.PayloadFormatVersion != RuntimeFlowCommandPayloadFormatVersionV2 {
		return nil, permanentPlatformCommandError(
			errors.New("runtime Flow command dependency executor is not configured for the command"),
		)
	}
	payload, inputs, err := decodeRuntimeFlowCommandPayloadV2(command.Payload)
	if err != nil {
		return nil, permanentPlatformCommandError(err)
	}
	return executeRuntimeFlowCommand(
		ctx,
		command,
		e.flows,
		e.users,
		runtimeFlowCommandPayload{
			definitionRef:     payload.DefinitionRef,
			definitionVersion: payload.DefinitionVersion,
			canonicalInputs:   payload.CanonicalInputs,
			dependencySource:  &payload.DependencySource,
		},
		inputs,
		true,
	)
}

func executeRuntimeFlowCommand(
	ctx context.Context,
	command PlatformCommandExecution,
	flows runtimeFlowCommandEngine,
	users runtimeFlowCommandSystemUserLoader,
	payload runtimeFlowCommandPayload,
	inputs map[string]interface{},
	allowDependency bool,
) ([]byte, error) {
	outcome, connectorNamespace, err := invokeRuntimeFlowCommand(
		ctx,
		command,
		flows,
		users,
		payload.definitionRef,
		payload.definitionVersion,
		inputs,
		allowDependency,
	)
	var recoveryRequested *automation.ConnectorInboundRemoteIdentityRecoveryRequestedError
	if errors.As(err, &recoveryRequested) {
		if !allowDependency {
			return nil, permanentPlatformCommandError(
				errors.New("runtime Flow command recovery request requires the V2 durable command profile"),
			)
		}
		return nil, &PlatformCommandRecoveryError{
			Request: automation.ConnectorInboundRemoteIdentityRecoveryRequest{
				Selector: recoveryRequested.Selector,
			},
		}
	}
	if outcome != nil && outcome.Dependency != nil && allowDependency {
		if payload.dependencySource == nil {
			return nil, permanentPlatformCommandError(
				errors.New("runtime Flow command dependency source is missing"),
			)
		}
		dependency, bindErr := contractmodels.BindFlowIdentityConvergenceDependency(
			*outcome.Dependency,
			connectorNamespace,
			payload.dependencySource.ObjectKind,
			payload.dependencySource.SourceID,
		)
		if bindErr != nil {
			return nil, permanentPlatformCommandError(
				fmt.Errorf("runtime Flow command dependency is invalid: %w", bindErr),
			)
		}
		return nil, &PlatformCommandDependencyError{Dependency: dependency}
	}
	if outcome != nil && outcome.Result != nil &&
		(outcome.Dependency != nil || !outcome.Result.IsSuccess ||
			outcome.Result.IsRetriable) {
		return nil, classifyRuntimeFlowCommandFailure(outcome)
	}
	if err != nil {
		return nil, err
	}
	if outcome == nil || outcome.Result == nil {
		return nil, permanentPlatformCommandError(
			errors.New("runtime Flow command returned no execution outcome"),
		)
	}
	if strings.TrimSpace(outcome.ExecutionID) == "" {
		return nil, permanentPlatformCommandError(
			errors.New("runtime Flow command returned no execution identity"),
		)
	}
	outputs, err := canonicalRuntimeFlowCommandJSON(
		outcome.Result.OutputValues,
	)
	if err != nil {
		return nil, permanentPlatformCommandError(
			fmt.Errorf("runtime Flow command output is not canonical JSON: %w", err),
		)
	}
	result, err := canonicalRuntimeFlowCommandJSON(runtimeFlowCommandResult{
		ExecutionID:      outcome.ExecutionID,
		CanonicalOutputs: outputs,
	})
	if err != nil {
		return nil, permanentPlatformCommandError(
			fmt.Errorf("runtime Flow command result is not canonical JSON: %w", err),
		)
	}
	return result, nil
}

// invokeRuntimeFlowCommand is the single immutable Flow selection, actor, and
// execution path shared by every durable Flow payload version. Callers retain
// version-specific dependency and terminal-result semantics.
func invokeRuntimeFlowCommand(
	ctx context.Context,
	command PlatformCommandExecution,
	flows runtimeFlowCommandEngine,
	users runtimeFlowCommandSystemUserLoader,
	definitionRef string,
	definitionVersion string,
	inputs map[string]interface{},
	allowDependency bool,
) (*contractmodels.FlowExecutionTypedOutcome, string, error) {
	if ctx == nil {
		return nil, "", errors.New("runtime Flow command context is required")
	}
	if err := ctx.Err(); err != nil {
		return nil, "", err
	}
	if flows == nil || users == nil || command.CommandKind != RuntimeFlowCommandKind {
		return nil, "", permanentPlatformCommandError(
			errors.New("runtime Flow command executor is not configured for the command"),
		)
	}
	if command.DefinitionAuthority == nil {
		return nil, "", permanentPlatformCommandError(
			errors.New("runtime Flow command payload is outside immutable command authority"),
		)
	}
	connectorNamespace, authorized := command.DefinitionAuthority.CommandExecutionContext(
		command.CommandKind,
		command.PayloadFormatVersion,
		command.CallerWorkloadID,
		definitionRef,
		definitionVersion,
	)
	if !authorized || strings.TrimSpace(connectorNamespace) == "" {
		return nil, "", permanentPlatformCommandError(
			errors.New("runtime Flow command payload is outside immutable command authority"),
		)
	}
	flowView, ok := runtimeFlowMetadataViewFromContext(ctx)
	if !ok || flowView == nil {
		return nil, "", permanentPlatformCommandError(
			errors.New("runtime Flow command is missing its immutable Flow view"),
		)
	}
	selector := flowexecution.SelectPlatformCommand
	if allowDependency {
		selector = flowexecution.SelectPlatformCommandWithDependency
	}
	selection, err := selector(
		ctx,
		flowView,
		definitionRef,
		flows.ValidateAtomicFlowForExecution,
	)
	if err != nil {
		if contextErr := ctx.Err(); contextErr != nil {
			return nil, "", contextErr
		}
		return nil, "", permanentPlatformCommandError(err)
	}
	if selection.Root == nil || selection.Root.Id != definitionVersion {
		return nil, "", permanentPlatformCommandError(
			errors.New("runtime Flow command definition version is not the exact selected version"),
		)
	}
	prepared := make(map[string]*models.FlowVersion, len(selection.Versions))
	for _, version := range selection.Versions {
		prepared[version.DefinitionRef] = version.Version
	}
	user, err := users.LoadSystemUserSession(ctx)
	if err != nil {
		return nil, "", err
	}
	if user == nil || user.ID != constants.SystemUserID {
		return nil, "", permanentPlatformCommandError(
			errors.New("runtime Flow command system actor is invalid"),
		)
	}
	executionCtx := constants.WithSecurityContext(
		ctx,
		constants.SystemContext,
		"runtime platform command Flow execution",
	)
	executionCtx = context.WithValue(
		executionCtx,
		constants.ContextKeyConnectorSyncOrigin,
		connectorNamespace,
	)
	// The command projection already binds this workload to the exact root name
	// and version. connectorNamespace is the Integration binding owner; it may
	// intentionally differ from the owner of an L2 companion Flow. Preserve both
	// connector execution markers for action semantics, but do not replace the
	// immutable command/Flow authority with a same-pack comparison here.
	executionCtx = constants.WithConnectorInboundFlowExecution(
		executionCtx,
		definitionRef,
		connectorNamespace,
	)
	executionCtx = constants.WithConnectorInboundCompanionFlowExecution(
		executionCtx,
		definitionVersion,
	)
	outcome, err := flows.ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		executionCtx,
		definitionRef,
		prepared,
		inputs,
		user,
		selection.MaxElementExecutions,
	)
	if contextErr := executionCtx.Err(); contextErr != nil {
		return nil, "", contextErr
	}
	return outcome, connectorNamespace, err
}

func classifyRuntimeFlowCommandFailure(
	outcome *contractmodels.FlowExecutionTypedOutcome,
) error {
	if outcome.Dependency != nil {
		return permanentPlatformCommandError(
			errors.New(
				"runtime Flow command profile does not support deferred dependency handoff",
			),
		)
	}
	cause := errors.New("runtime Flow command returned a failed result")
	permanent := !outcome.Result.IsRetriable
	if outcome.Cause != nil {
		cause = fmt.Errorf(
			"runtime Flow command failed: %s",
			outcome.Cause.Category,
		)
		permanent = !outcome.Cause.Retriable
	}
	// V1/V2 have no application-result envelope. Preserve the authored business
	// rejection code in their terminal evidence without persisting its message,
	// inputs, or output values. This does not reinterpret the retry taxonomy.
	if outcome.Result.IsCustomError && len(outcome.Result.Errors) > 0 {
		code := outcome.Result.Errors[0].StatusCode
		if code != "" && len(code) <= 128 && strings.IndexFunc(code, func(r rune) bool {
			return !(r >= 'A' && r <= 'Z' || r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || r == '_')
		}) == -1 {
			cause = fmt.Errorf("runtime Flow command failed: %s", code)
		}
	}
	return &PlatformCommandExecutionError{
		Cause: cause, Permanent: permanent,
	}
}

func permanentPlatformCommandError(cause error) error {
	return &PlatformCommandExecutionError{Cause: cause, Permanent: true}
}

func decodeRuntimeFlowCommandPayload(
	raw []byte,
) (flowcommandsemantic.Payload, map[string]interface{}, error) {
	payload, err := flowcommandsemantic.Decode(raw)
	if err != nil {
		return flowcommandsemantic.Payload{}, nil, err
	}
	inputs, err := decodeRuntimeFlowCommandInputs(payload.CanonicalInputs)
	if err != nil {
		return flowcommandsemantic.Payload{}, nil, err
	}
	return payload, inputs, nil
}

func decodeRuntimeFlowCommandPayloadV2(
	raw []byte,
) (flowcommandsemanticv2.Payload, map[string]interface{}, error) {
	payload, err := flowcommandsemanticv2.Decode(raw)
	if err != nil {
		return flowcommandsemanticv2.Payload{}, nil, err
	}
	inputs, err := decodeRuntimeFlowCommandInputs(payload.CanonicalInputs)
	if err != nil {
		return flowcommandsemanticv2.Payload{}, nil, err
	}
	return payload, inputs, nil
}

func decodeRuntimeFlowCommandInputs(
	canonicalInputs []byte,
) (map[string]interface{}, error) {
	inputs := map[string]interface{}{}
	inputDecoder := json.NewDecoder(strings.NewReader(string(canonicalInputs)))
	inputDecoder.UseNumber()
	if err := inputDecoder.Decode(&inputs); err != nil {
		return nil, errors.New("runtime Flow command inputs are invalid")
	}
	return inputs, nil
}

func canonicalRuntimeFlowCommandJSON(value interface{}) ([]byte, error) {
	if value == nil {
		value = map[string]interface{}{}
	}
	raw, err := json.Marshal(value)
	if err != nil {
		return nil, err
	}
	canonical, err := mutationsemantic.CanonicalSourceJSON(string(raw))
	if err != nil {
		return nil, err
	}
	return []byte(canonical), nil
}

var _ PlatformCommandExecutor = (*runtimeFlowCommandExecutor)(nil)
var _ PlatformCommandExecutor = (*runtimeFlowCommandDependencyExecutor)(nil)
var _ runtimeFlowCommandEngine = (*automation.FlowExecutor)(nil)