package ngoperation

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/devops/api"
	opent "github.com/tidbcloud/devops/entity/operation"
	"github.com/tidbcloud/devops/pkg/infra/clients/mgmthttpcli"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	"github.com/tidbcloud/devops/pkg/service/operation/actions"
	"github.com/tidbcloud/devops/pkg/service/operation/repo"
)

var (
	ActionReplaceNodeForNextGenResourcePool = opent.ActionMetadata{
		Name: "ReplaceNodeForNextGenResourcePool",
		Type: opent.ActionTypeAuto,
	}
)

const (
	replaceNodePhasePreparing = "REPLACE_NODE_PHASE_PREPARING"
	replaceNodePhasePrepared  = "REPLACE_NODE_PHASE_PREPARED"
	replaceNodePhaseCommitted = "REPLACE_NODE_PHASE_COMMITTED"
	replaceNodePhaseSucceeded = "REPLACE_NODE_PHASE_SUCCEEDED"
	replaceNodePhaseAborted   = "REPLACE_NODE_PHASE_ABORTED"
	replaceNodePhaseFailed    = "REPLACE_NODE_PHASE_FAILED"

	replaceNodePrepareTimeout = 15 * time.Minute
	replaceNodeCommitTimeout  = 15 * time.Minute
)

const (
	outputReplaceNodePhase            opent.OutputName = "replace_node_phase"
	outputReplaceNodePrepareStartedAt opent.OutputName = "replace_node_prepare_started_at"
	outputReplaceNodeCommitStartedAt  opent.OutputName = "replace_node_commit_started_at"
	outputReplaceNodePodName          opent.OutputName = "replace_node_pod_name"
	outputReplaceNodeInstanceID       opent.OutputName = "replace_node_instance_id"
	outputReplaceNodeOldNode          opent.OutputName = "replace_node_old_node"
	outputReplaceNodeWarmUpPod        opent.OutputName = "replace_node_warm_up_pod"
	outputReplaceNodeWarmUpNode       opent.OutputName = "replace_node_warm_up_node"
	outputReplaceNodeReplacementPod   opent.OutputName = "replace_node_replacement_pod"
	outputReplaceNodeReplacementNode  opent.OutputName = "replace_node_replacement_node"
	outputReplaceNodeMessage          opent.OutputName = "replace_node_message"
)

var _ actions.Action = &replaceNode{}

type replaceNode struct {
	actions.BaseAction
	comps *actions.Components
}

func NewReplaceNodeForNextGenResourcePool(comps *actions.Components) actions.AutoAction {
	return &replaceNode{comps: comps}
}

func (a *replaceNode) Metadata() opent.ActionMetadata {
	return ActionReplaceNodeForNextGenResourcePool
}

func (a *replaceNode) PreCheck(ctx context.Context, instance *opent.Instance) error {
	if err := a.checkResourcePoolExistence(ctx, instance); err != nil {
		return err
	}

	if _, err := a.getParam(instance); err != nil {
		return err
	}

	return a.ensureNoRunningConflict(ctx, instance)
}

func (a *replaceNode) Start(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	if err := a.checkResourcePoolExistence(ctx, instance); err != nil {
		return actions.FailedActionResult(err), nil
	}

	req, err := a.getParam(instance)
	if err != nil {
		return actions.FailedActionResult(err), nil
	}

	agent := a.replaceNodeAgent(req)
	state := newReplaceNodeState(nil)
	resp, err := agent.PrepareReplaceNode(ctx, req)
	if err != nil {
		return abortBeforeCommit(ctx, agent, req, state,
			fmt.Errorf("prepare replace node failed: %w", err),
			func(abortErr error) error {
				return fmt.Errorf("prepare replace node failed: %w; abort also failed: %s", err, abortErr)
			},
			true,
		)
	}

	state.merge(resp)
	state.markPrepareStarted(time.Now())
	return actions.RunningActionResult(replaceNodeActionOutput("prepare", req, resp), state.outputs), nil
}

func (a *replaceNode) Result(ctx context.Context, instance *opent.Instance) (*actions.ActionResult, error) {
	state := newReplaceNodeState(instance.Outputs)
	if err := a.checkResourcePoolExistence(ctx, instance); err != nil {
		if state.hasCommitMarker() {
			commitStartedAt := state.commitStartedAt(time.Now())
			if time.Since(commitStartedAt) > replaceNodeCommitTimeout {
				return failedReplaceNodeActionResult(fmt.Errorf("resource pool existence check failed after commit timeout %s: %w. Do not run abort automatically; inspect cluster-agent logs, operator instance conditions, replacement pod scheduling, and old node cordon state", replaceNodeCommitTimeout, err), state.outputs), nil
			}
			return actions.RunningActionResult(fmt.Sprintf("resource pool existence check failed after commit started, retrying until commit timeout: %s", err), state.outputs), nil
		}
		return actions.NewActionResult(opent.ActionStatusUnknown, err.Error(), nil), nil
	}

	req, err := a.getParam(instance)
	if err != nil {
		return actions.FailedActionResult(err), nil
	}

	agent := a.replaceNodeAgent(req)
	prepareStartedAt := state.prepareStartedAt(time.Now())

	resp, err := agent.GetReplaceNodeOperation(ctx, replaceNodeOperationReq(req))
	if err == nil && resp == nil {
		err = errors.New("get replace node operation returned empty response")
	}
	if err != nil {
		if state.hasCommitMarker() {
			commitStartedAt := state.commitStartedAt(time.Now())
			if time.Since(commitStartedAt) > replaceNodeCommitTimeout {
				return failedReplaceNodeActionResult(fmt.Errorf("get replace node operation failed after commit timeout %s: %w. Do not run abort automatically; inspect cluster-agent logs, operator instance conditions, replacement pod scheduling, and old node cordon state", replaceNodeCommitTimeout, err), state.outputs), nil
			}
			return actions.RunningActionResult(fmt.Sprintf("get replace node operation failed after commit started, retrying until commit timeout: %s", err), state.outputs), nil
		}
		if time.Since(prepareStartedAt) > replaceNodePrepareTimeout {
			return abortBeforeCommit(ctx, agent, req, state,
				fmt.Errorf("get replace node operation timed out after %s and abort completed: %w", replaceNodePrepareTimeout, err),
				func(abortErr error) error {
					return fmt.Errorf("get replace node operation timed out after %s: %w; abort also failed: %s", replaceNodePrepareTimeout, err, abortErr)
				},
				true,
			)
		}
		return actions.RunningActionResult(fmt.Sprintf("get replace node operation failed before commit, retrying until prepare timeout: %s", err), state.outputs), nil
	}
	state.merge(resp)

	if result, ok := terminalReplaceNodeResult(resp, state.outputs); ok {
		return result, nil
	}

	if state.hasCommitMarker() {
		return a.pollReplaceNodeCommit(ctx, instance, req, resp, state)
	}

	if resp.Phase == replaceNodePhaseCommitted {
		return a.recoverReplaceNodeCommitMarker(ctx, instance, req, resp, state)
	}

	if isReplaceNodePreparedForCommit(resp.Phase) {
		return a.startReplaceNodeCommit(ctx, instance, agent, req, resp, state)
	}

	if time.Since(prepareStartedAt) > replaceNodePrepareTimeout {
		return abortBeforeCommit(ctx, agent, req, state,
			fmt.Errorf("prepare replace node timed out after %s; abort completed", replaceNodePrepareTimeout),
			func(abortErr error) error {
				return fmt.Errorf("prepare replace node timed out after %s; abort failed: %w", replaceNodePrepareTimeout, abortErr)
			},
			false,
		)
	}

	return actions.RunningActionResult(replaceNodeActionOutput("prepare", req, resp), state.outputs), nil
}

func (a *replaceNode) OnCancel(ctx context.Context, instance *opent.Instance) (bool, error) {
	if newReplaceNodeState(instance.Outputs).hasCommitMarker() {
		return false, nil
	}

	if err := a.checkResourcePoolExistence(ctx, instance); err != nil {
		return false, err
	}
	req, err := a.getParam(instance)
	if err != nil {
		return false, err
	}

	_, err = a.replaceNodeAgent(req).AbortReplaceNode(ctx, replaceNodeOperationReq(req))
	if err != nil {
		if isReplaceNodeAbortRejectedAfterCommitError(err) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (a *replaceNode) startReplaceNodeCommit(ctx context.Context, instance *opent.Instance, agent mgmthttpcli.NextGenClusterAgent, req *mgmthttpcli.PrepareReplaceNodeReq, prepareResp *mgmthttpcli.ReplaceNodeResp, state *replaceNodeState) (*actions.ActionResult, error) {
	state.markCommitStarted(time.Now())
	if err := a.persistReplaceNodeCommitMarker(instance, state.outputs); err != nil {
		return nil, err
	}

	resp, err := agent.CommitReplaceNode(ctx, replaceNodeOperationReq(req))
	if err != nil {
		return failedReplaceNodeActionResult(fmt.Errorf("commit replace node failed after prepare phase %s: %w. Do not run abort automatically; inspect cluster-agent logs, operator instance conditions, replacement pod scheduling, and old node cordon state", prepareResp.Phase, err), state.outputs), nil
	}
	state.merge(resp)
	if result, ok := terminalReplaceNodeResult(resp, state.outputs); ok {
		return result, nil
	}
	return actions.RunningActionResult(replaceNodeActionOutput("commit", req, resp), state.outputs), nil
}

func (a *replaceNode) recoverReplaceNodeCommitMarker(ctx context.Context, instance *opent.Instance, req *mgmthttpcli.PrepareReplaceNodeReq, resp *mgmthttpcli.ReplaceNodeResp, state *replaceNodeState) (*actions.ActionResult, error) {
	state.markCommitStarted(time.Now())
	if err := a.persistReplaceNodeCommitMarker(instance, state.outputs); err != nil {
		return nil, err
	}
	return actions.RunningActionResult(replaceNodeActionOutput("commit", req, resp), state.outputs), nil
}

func (a *replaceNode) pollReplaceNodeCommit(ctx context.Context, instance *opent.Instance, req *mgmthttpcli.PrepareReplaceNodeReq, resp *mgmthttpcli.ReplaceNodeResp, state *replaceNodeState) (*actions.ActionResult, error) {
	commitStartedAt := state.commitStartedAt(time.Now())
	if result, ok := terminalReplaceNodeResult(resp, state.outputs); ok {
		return result, nil
	}
	if time.Since(commitStartedAt) > replaceNodeCommitTimeout {
		return failedReplaceNodeActionResult(fmt.Errorf("commit replace node timed out after %s. Do not run abort automatically; inspect cluster-agent logs, operator instance conditions, replacement pod scheduling, and old node cordon state", replaceNodeCommitTimeout), state.outputs), nil
	}
	if err := a.persistReplaceNodeOutputs(ctx, instance, state.outputs); err != nil {
		return nil, err
	}
	return actions.RunningActionResult(replaceNodeActionOutput("commit", req, resp), state.outputs), nil
}

type replaceNodeState struct {
	outputs map[opent.OutputName]string
}

func newReplaceNodeState(outputs map[opent.OutputName]string) *replaceNodeState {
	return &replaceNodeState{outputs: copyReplaceNodeOutputs(outputs)}
}

func (s *replaceNodeState) markPrepareStarted(now time.Time) {
	s.outputs[outputReplaceNodePrepareStartedAt] = strconv.FormatInt(now.Unix(), 10)
}

func (s *replaceNodeState) markCommitStarted(now time.Time) {
	s.outputs[outputReplaceNodeCommitStartedAt] = strconv.FormatInt(now.Unix(), 10)
}

func (s *replaceNodeState) prepareStartedAt(fallback time.Time) time.Time {
	return ensureReplaceNodeOutputTime(s.outputs, outputReplaceNodePrepareStartedAt, fallback)
}

func (s *replaceNodeState) commitStartedAt(fallback time.Time) time.Time {
	return ensureReplaceNodeOutputTime(s.outputs, outputReplaceNodeCommitStartedAt, fallback)
}

func (s *replaceNodeState) hasCommitMarker() bool {
	return s.outputs[outputReplaceNodeCommitStartedAt] != ""
}

func (s *replaceNodeState) merge(resp *mgmthttpcli.ReplaceNodeResp) {
	mergeReplaceNodeOutputs(s.outputs, replaceNodeOutputs(resp))
}

func terminalReplaceNodeResult(resp *mgmthttpcli.ReplaceNodeResp, outputs map[opent.OutputName]string) (*actions.ActionResult, bool) {
	if resp == nil {
		return nil, false
	}
	switch resp.Phase {
	case replaceNodePhaseSucceeded:
		return actions.SucceededActionResult(outputs), true
	case replaceNodePhaseAborted:
		return failedReplaceNodeActionResult(fmt.Errorf("replace node operation was aborted before commit: %s", replaceNodeResolvedSummary(resp)), outputs), true
	case replaceNodePhaseFailed:
		return failedReplaceNodeActionResult(fmt.Errorf("replace node operation failed: %s. Inspect cluster-agent logs, operator instance conditions, replacement pod scheduling, and old node cordon state", replaceNodeResolvedSummary(resp)), outputs), true
	default:
		return nil, false
	}
}

func failedReplaceNodeActionResult(err error, outputs map[opent.OutputName]string) *actions.ActionResult {
	return actions.NewActionResult(opent.ActionStatusFailed, err.Error(), outputs)
}

func abortBeforeCommit(
	ctx context.Context,
	agent mgmthttpcli.NextGenClusterAgent,
	req *mgmthttpcli.PrepareReplaceNodeReq,
	state *replaceNodeState,
	completedErr error,
	abortErrFn func(error) error,
	ignoreNotFound bool,
) (*actions.ActionResult, error) {
	abortResp, abortErr := agent.AbortReplaceNode(ctx, replaceNodeOperationReq(req))
	if abortErr != nil && (!ignoreNotFound || !isReplaceNodeAbortNotFoundError(abortErr)) {
		return failedReplaceNodeActionResult(abortErrFn(abortErr), state.outputs), nil
	}
	state.merge(abortResp)
	return failedReplaceNodeActionResult(completedErr, state.outputs), nil
}

func (a *replaceNode) getParam(instance *opent.Instance) (*mgmthttpcli.PrepareReplaceNodeReq, error) {
	resourcePoolID := strings.TrimSpace(instance.Params[opent.ParamResourcePoolID])
	if resourcePoolID == "" {
		return nil, fmt.Errorf("resource_pool_id is required")
	}

	podName := strings.TrimSpace(instance.Params[opent.ParamPodName])
	instanceID := strings.TrimSpace(instance.Params[opent.ParamInstanceID])
	if (podName == "") == (instanceID == "") {
		return nil, fmt.Errorf("exactly one of pod_name or instance_id is required")
	}

	return &mgmthttpcli.PrepareReplaceNodeReq{
		OperationID:    strconv.FormatUint(instance.ID, 10),
		ResourcePoolID: resourcePoolID,
		PodName:        podName,
		InstanceID:     instanceID,
		Force:          instance.Params[opent.ParamForceReplace] == "true",
	}, nil
}

func replaceNodeOperationReq(req *mgmthttpcli.PrepareReplaceNodeReq) *mgmthttpcli.ReplaceNodeOperationReq {
	return &mgmthttpcli.ReplaceNodeOperationReq{
		OperationID:    req.OperationID,
		ResourcePoolID: req.ResourcePoolID,
	}
}

func (a *replaceNode) checkResourcePoolExistence(ctx context.Context, instance *opent.Instance) error {
	resourcePoolID, err := actions.GetResourcePoolID(instance)
	if err != nil {
		return err
	}
	_, err = a.comps.CoreSvc.Mgmt.GetResourcePoolString(ctx, resourcePoolID)
	return err
}

func (a *replaceNode) replaceNodeAgent(req *mgmthttpcli.PrepareReplaceNodeReq) mgmthttpcli.NextGenClusterAgent {
	return a.comps.MgmtHTTP.NextGenClusterAgent("", req.ResourcePoolID)
}

func (a *replaceNode) ensureNoRunningConflict(ctx context.Context, instance *opent.Instance) error {
	if a.comps == nil {
		return nil
	}
	if a.comps.OpRepo == nil {
		a.comps.Logger.Info("skip replace node running conflict check because operation repo is nil", "instance", instance.ID, "operation", instance.Operation)
		return nil
	}

	resourcePoolID, err := actions.GetResourcePoolID(instance)
	if err != nil {
		return err
	}

	instances, err := a.comps.OpRepo.ListInstances(ctx,
		repo.InstanceOpts.TypeIs(opent.OperationTargetResourcePool),
		repo.InstanceOpts.TargetIDIn(opent.OperationTargetResourcePool, []string{resourcePoolID}),
		repo.InstanceOpts.OperationIn([]string{opent.OperationTypeReplaceNodeForNextGenResourcePool}),
		repo.InstanceOpts.PhaseNotIn(opent.InstancePhaseFinishedValues),
	)
	if err != nil {
		return err
	}

	for _, existing := range instances {
		if existing.ID == instance.ID {
			continue
		}
		return fmt.Errorf("another ReplaceNodeForNextGenResourcePool operation %d is already in progress for resource pool %s", existing.ID, resourcePoolID)
	}
	return nil
}

func (a *replaceNode) persistReplaceNodeOutputs(ctx context.Context, instance *opent.Instance, outputs map[opent.OutputName]string) error {
	if a.comps == nil || a.comps.OpRepo == nil {
		if instance.Outputs == nil {
			instance.Outputs = map[opent.OutputName]string{}
		}
		mergeReplaceNodeOutputs(instance.Outputs, outputs)
		return nil
	}
	if err := a.comps.OpRepo.PatchInstance(ctx, instance, func(current *opent.Instance) error {
		if current.Outputs == nil {
			current.Outputs = map[opent.OutputName]string{}
		}
		mergeReplaceNodeOutputs(current.Outputs, outputs)
		return nil
	}); err != nil {
		return err
	}
	if instance.Outputs == nil {
		instance.Outputs = map[opent.OutputName]string{}
	}
	mergeReplaceNodeOutputs(instance.Outputs, outputs)
	return nil
}

func (a *replaceNode) persistReplaceNodeCommitMarker(instance *opent.Instance, outputs map[opent.OutputName]string) error {
	// Result runs inside a transaction that is rolled back when the action stays Running.
	// The pre-commit marker must survive that retry rollback before the external commit RPC.
	return a.persistReplaceNodeOutputs(context.Background(), instance, outputs)
}

func isReplaceNodePreparedForCommit(phase string) bool {
	return phase == replaceNodePhasePrepared
}

func isReplaceNodeAbortNotFoundError(err error) bool {
	if err == nil {
		return false
	}
	return errs.IsNotFound(err)
}

func isReplaceNodeAbortRejectedAfterCommitError(err error) bool {
	if err == nil {
		return false
	}
	if apiErr, ok := errs.AsAPIError(err); ok && apiErr.ErrCode() == api.ErrCodeFailedPrecond {
		return true
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "committed") ||
		strings.Contains(msg, "failedprecondition") ||
		strings.Contains(msg, "failed_precondition") ||
		strings.Contains(msg, "precondition failed") ||
		strings.Contains(msg, "412")
}

func ensureReplaceNodeOutputTime(outputs map[opent.OutputName]string, name opent.OutputName, fallback time.Time) time.Time {
	value := outputs[name]
	if value == "" {
		outputs[name] = strconv.FormatInt(fallback.Unix(), 10)
		return fallback
	}
	seconds, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		outputs[name] = strconv.FormatInt(fallback.Unix(), 10)
		return fallback
	}
	return time.Unix(seconds, 0)
}

func copyReplaceNodeOutputs(in map[opent.OutputName]string) map[opent.OutputName]string {
	out := map[opent.OutputName]string{}
	for key, value := range in {
		out[key] = value
	}
	return out
}

func replaceNodeOutputs(resp *mgmthttpcli.ReplaceNodeResp) map[opent.OutputName]string {
	outputs := map[opent.OutputName]string{}
	if resp == nil {
		return outputs
	}
	setReplaceNodeOutput(outputs, outputReplaceNodePhase, resp.Phase)
	setReplaceNodeOutput(outputs, outputReplaceNodePodName, resp.PodName)
	setReplaceNodeOutput(outputs, outputReplaceNodeInstanceID, resp.InstanceID)
	setReplaceNodeOutput(outputs, outputReplaceNodeOldNode, resp.OldNode)
	setReplaceNodeOutput(outputs, outputReplaceNodeWarmUpPod, resp.WarmUpPod)
	setReplaceNodeOutput(outputs, outputReplaceNodeWarmUpNode, resp.WarmUpNode)
	setReplaceNodeOutput(outputs, outputReplaceNodeReplacementPod, resp.ReplacementPod)
	setReplaceNodeOutput(outputs, outputReplaceNodeReplacementNode, resp.ReplacementNode)
	setReplaceNodeOutput(outputs, outputReplaceNodeMessage, resp.Message)
	return outputs
}

func setReplaceNodeOutput(outputs map[opent.OutputName]string, name opent.OutputName, value string) {
	if value != "" {
		outputs[name] = value
	}
}

func mergeReplaceNodeOutputs(dst, src map[opent.OutputName]string) {
	if dst == nil {
		return
	}
	for key, value := range src {
		if value != "" {
			dst[key] = value
		}
	}
}

func replaceNodeActionOutput(stage string, req *mgmthttpcli.PrepareReplaceNodeReq, resp *mgmthttpcli.ReplaceNodeResp) string {
	return fmt.Sprintf("%s replace node for %s: %s", stage, replaceNodeTargetSummary(req), replaceNodeResolvedSummary(resp))
}

func replaceNodeTargetSummary(req *mgmthttpcli.PrepareReplaceNodeReq) string {
	if req.PodName != "" {
		return fmt.Sprintf("pod_name=%s", req.PodName)
	}
	return fmt.Sprintf("instance_id=%s", req.InstanceID)
}

func replaceNodeResolvedSummary(resp *mgmthttpcli.ReplaceNodeResp) string {
	if resp == nil {
		return "accepted"
	}

	parts := make([]string, 0, 8)
	if resp.Phase != "" {
		parts = append(parts, fmt.Sprintf("phase=%s", resp.Phase))
	}
	if resp.PodName != "" {
		parts = append(parts, fmt.Sprintf("pod_name=%s", resp.PodName))
	}
	if resp.ComponentKind != "" {
		parts = append(parts, fmt.Sprintf("component_kind=%s", resp.ComponentKind))
	}
	if resp.OldNode != "" {
		parts = append(parts, fmt.Sprintf("old_node=%s", resp.OldNode))
	}
	if resp.WarmUpNode != "" {
		parts = append(parts, fmt.Sprintf("warm_up_node=%s", resp.WarmUpNode))
	}
	if resp.ReplacementNode != "" {
		parts = append(parts, fmt.Sprintf("replacement_node=%s", resp.ReplacementNode))
	}
	if resp.InstanceID != "" {
		parts = append(parts, fmt.Sprintf("instance_id=%s", resp.InstanceID))
	}
	if resp.Message != "" {
		parts = append(parts, fmt.Sprintf("message=%s", resp.Message))
	}
	if len(parts) == 0 {
		return "accepted"
	}
	return strings.Join(parts, ", ")
}