package dataplane

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

	"github.com/mitchellh/mapstructure"
	"github.com/samber/lo"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	clusteragent "github.com/tidbcloud/eks-provider/controllers/utils/cluster_agent"
	"github.com/tidbcloud/eks-provider/controllers/workflow/action"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
)

// togglePDScheduler adds or removes a PD scheduler via cluster-agent pd-ctl
type togglePDScheduler struct {
	cli client.Client
}

type togglePDSchedulerInputs struct {
	Namespace   string `mapstructure:"namespace" json:"namespace" validate:"required"`
	ClusterName string `mapstructure:"clusterName" json:"clusterName" validate:"required"`
	Scheduler   string `mapstructure:"scheduler" json:"scheduler" validate:"required"`
	// Op must be "add" or "remove"
	Op string `mapstructure:"op" json:"op" validate:"required,oneof=add remove"`
}

var (
	_ action.Action = &togglePDScheduler{}
)

func NewTogglePDScheduler(cli client.Client) action.Action {
	return &togglePDScheduler{cli: cli}
}

func (a *togglePDScheduler) Name() cloudv1.WorkflowActionName {
	return consts.WfActionTogglePDScheduler
}

func (a *togglePDScheduler) PreCheck(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	inputs := &togglePDSchedulerInputs{}
	if err := mapstructure.Decode(step.Inputs.Object, inputs); err != nil {
		return fmt.Errorf("decode inputs: %w", err)
	}
	if err := validate.Struct(inputs); err != nil {
		return fmt.Errorf("validate inputs: %w", err)
	}
	return nil
}

func (a *togglePDScheduler) Handle(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	logger := log.FromContext(ctx)
	inputs := &togglePDSchedulerInputs{}
	lo.Must0(mapstructure.Decode(step.Inputs.Object, inputs))

	// Fetch cluster to initialize cluster-agent client
	cluster, err := getCluster(ctx, a.cli, client.ObjectKey{Namespace: inputs.Namespace, Name: inputs.ClusterName})
	if err != nil {
		return err
	}
	caClient, err := clusteragent.NewClient(cluster)
	if err != nil {
		return fmt.Errorf("create cluster-agent client: %w", err)
	}

	// Discover PD endpoints via cluster-agent topology
	topo, err := caClient.GetClusterTopology(ctx, inputs.ClusterName)
	if err != nil {
		return fmt.Errorf("get cluster topology: %w", err)
	}
	if len(topo.PdNodes) == 0 {
		return fmt.Errorf("no PD nodes found for cluster %s", inputs.ClusterName)
	}
	pd := topo.PdNodes[0]
	pdURL := fmt.Sprintf("https://%s:%d", pd.Host, pd.Port)
	logger.Info("execute PD commands on", "url", pdURL)

	// Run pd-ctl to add/remove scheduler
	op := strings.ToLower(inputs.Op)
	cmd := fmt.Sprintf("pd-ctl -u %s scheduler %s %s", pdURL, op, inputs.Scheduler)
	execResp, err := caClient.ExecCommand(ctx, &apiv1.ExecuteCommandReq{ClusterId: inputs.ClusterName, Command: cmd})
	if err != nil {
		return fmt.Errorf("execute %q: %w", cmd, err)
	}
	if execResp.ExitCode != nil && *execResp.ExitCode != 0 {
		return fmt.Errorf("execute %q failed: code=%d stderr=%s", cmd, *execResp.ExitCode, strings.TrimSpace(execResp.StderrMsg))
	}

	schedulers, err := pdShowSchedulers(ctx, caClient, inputs.ClusterName, pdURL)
	if err != nil {
		return fmt.Errorf("show schedulers: %w", err)
	}

	switch op {
	case consts.WfInputKeyTogglePDSchedulerOpAdd:
		if lo.Contains(schedulers, inputs.Scheduler) {
			return nil
		}
		return fmt.Errorf("scheduler %s not found after add", inputs.Scheduler)
	case consts.WfInputKeyTogglePDSchedulerOpRemove:
		if !lo.Contains(schedulers, inputs.Scheduler) {
			return nil
		}
		return fmt.Errorf("scheduler %s still found after remove", inputs.Scheduler)
	default:
		return cloudv1.NewWFNonRetriableError(fmt.Errorf("unknown action %q", op))
	}
}

func pdShowSchedulers(ctx context.Context, caClient clusteragent.Client, clusterId string, pdURL string) ([]string, error) {
	logger := log.FromContext(ctx)
	showCmd := fmt.Sprintf("pd-ctl -u %s scheduler show", pdURL)
	descResp, err := caClient.ExecCommand(ctx, &apiv1.ExecuteCommandReq{ClusterId: clusterId, Command: showCmd})
	if err != nil {
		return nil, fmt.Errorf("execute %q: %w", showCmd, err)
	}
	if descResp.ExitCode != nil && *descResp.ExitCode != 0 {
		return nil, fmt.Errorf("execute %q failed: code=%d stderr=%s", showCmd, *descResp.ExitCode, strings.TrimSpace(descResp.StderrMsg))
	}

	var schedulers []string
	if err := json.Unmarshal([]byte(descResp.StdoutMsg), &schedulers); err != nil {
		return nil, fmt.Errorf("unmarshal pd-ctl scheduler show output: %w", err)
	}

	logger.Info("current PD schedulers", "schedulers", schedulers)
	return schedulers, nil
}