package clusterctl

import (
	"context"
	"encoding/json"
	"fmt"
	"regexp"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/devops/entity"
	auditent "github.com/tidbcloud/devops/entity/audit"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/clients/mgmthttpcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/serverlesscli"
	"github.com/tidbcloud/devops/pkg/infra/config"
	"github.com/tidbcloud/devops/pkg/infra/consts"
	"github.com/tidbcloud/devops/pkg/infra/utils/pointer"
	"github.com/tidbcloud/devops/pkg/service"
	"github.com/tidbcloud/devops/pkg/service/audit"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"

	clusteragent "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"

	"github.com/go-logr/logr"
	"go.uber.org/dig"
)

type ServiceParam struct {
	dig.In

	Conf          *config.RootConfig
	Logger        logr.Logger
	AuditAPI      audit.AuditServiceAPI
	CoreSvcCli    *coresvccli.CoreSvcCli
	MgmtHTTP      mgmthttpcli.MgmtHTTPClient
	ServerlessCli *serverlesscli.Client
}

type ServiceResult struct {
	dig.Out

	API     ClusterCtlServiceAPI
	Service service.Service `group:"service"`
}

type svc struct {
	*service.BaseService

	conf config.ClusterCtlServiceConfig

	coreCli       *coresvccli.CoreSvcCli
	auditSvc      audit.AuditServiceAPI
	mgmtHTTP      mgmthttpcli.MgmtHTTPClient
	serverlessCli *serverlesscli.Client
}

type AuditLogDetail struct {
	Command string `json:"command"`
}

func NewService(p ServiceParam) (ServiceResult, error) {
	base := service.NewBaseService("ClusterCtlService")
	s := &svc{
		BaseService:   base,
		conf:          p.Conf.Services.ClusterCtl,
		auditSvc:      p.AuditAPI,
		coreCli:       p.CoreSvcCli,
		mgmtHTTP:      p.MgmtHTTP,
		serverlessCli: p.ServerlessCli,
	}

	return ServiceResult{API: s, Service: s}, nil
}

func (s *svc) GetTopology(ctx context.Context, clusterID string) (entity.Topology, error) {
	var topo entity.Topology

	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return topo, fmt.Errorf("cluster_id is not a int")
	}

	cluster, err := s.coreCli.Mgmt.GetCluster(ctx, clusterIDUint64)
	if err != nil {
		return topo, err
	}
	topo.ClusterName = cluster.ClusterName

	req := &clusteragent.GetClusterTopologyReq{
		ClusterId: clusterID,
	}
	resp := &clusteragent.GetClusterTopologyResp{}
	switch cluster.DeployType {
	case mgmtv1.DeployType_DEPLOY_TYPE_DEDICATED:
		resp, err = s.mgmtHTTP.ClusterAgent(clusterID).GetClusterTopology(ctx, &clusteragent.GetClusterTopologyReq{
			ClusterId: clusterID,
		})
	case mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN:
		resp, err = s.mgmtHTTP.NextGenClusterAgent(clusterID, cluster.GetNextGenCluster().SharedPoolId).GetClusterTopology(ctx, req)
	case mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS:
		// Serverless shares a single PD/TiKV cluster across all serverless tenants.
		// pd-ctl is routed via ops-service exec into pd-ctl-runner pod, so the
		// Terminal does not need per-cluster PD/TiKV endpoints. Return the cluster
		// name only (already set above) with empty PD/TiKV lists. Initialize as
		// empty slices (not nil) so JSON marshals to [] instead of null.
		topo.PD = []entity.Instance{}
		topo.TiKV = []entity.Instance{}
		return topo, nil
	default:
		return topo, fmt.Errorf("unsupported deploy type: %s", cluster.DeployType.String())
	}
	if err != nil {
		return topo, fmt.Errorf("get cluster topology failed: %w", err)
	}

	for _, pd := range resp.GetPdNodes() {
		topo.PD = append(topo.PD, entity.Instance{
			Host: pd.GetHost(),
			Port: int(pd.GetPort()),
		})
	}
	for _, tikv := range resp.GetTikvNodes() {
		topo.TiKV = append(topo.TiKV, entity.Instance{
			Host: tikv.GetHost(),
			Port: int(tikv.GetPort()),
		})
	}

	return topo, nil
}

func (s *svc) ExecuteCtl(ctx context.Context, clusterID string, command string) (string, string, error) {
	// 1. check if the cluster exists
	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return "", "", fmt.Errorf("cluster_id is not a int")
	}

	cluster, err := s.coreCli.Mgmt.GetCluster(ctx, clusterIDUint64)
	if err != nil {
		return "", "", err
	}

	// 2. check if the command is blocked
	supportedPrograms := []string{"pd-ctl", "tikv-ctl"}
	if cluster.DeployType == mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN {
		supportedPrograms = []string{"pd-ctl", "cse-ctl"}
	}
	if cluster.DeployType == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS {
		supportedPrograms = []string{"pd-ctl"}
	}

	program, _ := splitCommand(command)
	if program == "" {
		return "", "", nil // ignore empty command
	}
	if !slices.Contains(supportedPrograms, program) {
		return "", "", fmt.Errorf("program %s is not supported", program)
	}
	if isBlockedCommand(s.conf.BlockedCommands, command) {
		return "", "", fmt.Errorf("command %s is blocked", command)
	}

	// 3. record audit log
	details, err := json.Marshal(AuditLogDetail{command})
	if err != nil {
		return "", "", err
	}
	s.auditSvc.Create(ctx, &auditent.AuditLog{
		Type:      auditent.AuditLogTypeCtl,
		ObjectID:  clusterID,
		Operation: "ctl",
		Operator:  consts.GetOperatorFromCtx(ctx, "unknown"),
		OperateAt: pointer.Pointer(time.Now()),
		Details:   details,
	})

	// 4. request to executor-server
	req := &clusteragent.ExecuteCommandReq{
		ClusterId: clusterID,
		Command:   command,
	}
	resp := &clusteragent.ExecuteCommandResp{}
	switch cluster.DeployType {
	case mgmtv1.DeployType_DEPLOY_TYPE_DEDICATED:
		resp, err = s.mgmtHTTP.ClusterAgent(clusterID).ExecuteCommand(ctx, req)
	case mgmtv1.DeployType_DEPLOY_TYPE_NEXTGEN:
		resp, err = s.mgmtHTTP.NextGenClusterAgent(clusterID, cluster.GetNextGenCluster().SharedPoolId).ExecuteCommand(ctx, req)
	case mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS:
		clusterDetails, sErr := s.serverlessCli.GetCluster(ctx,
			clusterID,
			strconv.FormatUint(cluster.ProjectId, 10),
			strconv.FormatUint(cluster.OrganizationId, 10),
		)
		if sErr != nil {
			return "", "", fmt.Errorf("get serverless cluster details: %w", sErr)
		}
		if clusterDetails.Kubernetes == "" {
			return "", "", fmt.Errorf("cluster %s has no kubernetes name", clusterID)
		}
		opsResp, sErr := s.serverlessCli.ExecuteCommand(ctx,
			clusterDetails.Provider,
			cluster.Region,
			clusterDetails.Kubernetes,
			command,
		)
		if sErr != nil {
			return "", "", fmt.Errorf("exec serverless pd-ctl: %w", sErr)
		}
		return opsResp.Stdout, opsResp.Stderr, nil
	default:
		return "", "", fmt.Errorf("unsupported deploy type: %s", cluster.DeployType.String())
	}
	if err != nil {
		return "", "", fmt.Errorf("exec ctl cmd failed: %w", err)
	}

	return resp.GetStdoutMsg(), resp.GetStderrMsg(), nil

}

func isBlockedCommand(rules []string, command string) bool {
	for _, rule := range rules {
		if regexp.MustCompile(rule).MatchString(command) {
			return true
		}
	}
	return false
}

func splitCommand(command string) (program, args string) {
	fields := strings.Fields(command)
	if len(fields) == 0 {
		return "", ""
	}

	program = fields[0]
	args = strings.TrimPrefix(command, fields[0])

	return program, args
}