package premium_dm

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"strconv"
	"strings"

	"github.com/pingcap/errors"
	dm "github.com/pingcap/tiflow/dm/openapi"
	"github.com/tidbcloud/dataflow-service-ng/pkg/agent"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	pkglog "github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/dataflow-service-ng/pkg/redact"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	agentpb "github.com/tidbcloud/dataflow-service/dataflow-agent-ng/idl/pbgen/proto/ng_agent"
	"go.uber.org/zap"
)

const (
	CodeSchedulerSourceCfgNotExist     = 46008
	CodeSchedulerSubTaskOpTaskNotExist = 46016
	CodeSchedulerTaskNotExist          = 46018
	CodeSchedulerSubTaskExist          = 46014
)

const (
	PremiumDMSourcePath = "/api/v1/sources"
	PremiumDMTaskPath   = "/api/v1/tasks"
	PremiumDMWorkerPath = "/api/v1/cluster/workers"

	DmWorkerPrefix = "dbcluster-dm-worker-"
)

type PremiumDMProxy struct {
	agentCli        *agent.NGRegionalClient
	dmOpenAPI       regionalapi.DMOpenAPIClient
	useServiceProxy bool
	dmClusterID     string
	gatewayBaseURL  string
	DMNameSpace     string
}

type dmRequestOpts struct {
	patchTargetConn bool
	clusterID       string
}

var (
	newNGRegionalClientByClusterIDFunc = agent.NewNGRegionalClientByClusterID
	getSharedPoolByClusterIDFunc       = func(
		ctx context.Context, agentClient *agent.NGRegionalClient, clusterID uint64,
	) (*agent.SharedPool, error) {
		return agentClient.GetSharePoolByClusterID(ctx, clusterID)
	}
	resolveDMClusterForTiDBIDFunc = ResolveDMClusterForTiDBID
	newDMOpenAPIClientFunc        = regionalapi.NewDMOpenAPIClient
)

func NewPremiumDMProxy(
	ctx context.Context,
	tiDBID string,
	mgmtClient grpcli.MgmtSvcGrpcClient,
	agentCfg agent.ClientConfig,
	dmCluster *agent.DMCluster,
) (*PremiumDMProxy, error) {
	clusterID, err := strconv.ParseUint(tiDBID, 10, 64)
	if err != nil {
		return nil, err
	}
	agentClient, err := newNGRegionalClientByClusterIDFunc(ctx, clusterID, mgmtClient, &agentCfg)
	if err != nil {
		return nil, err
	}
	sharedPool, err := getSharedPoolByClusterIDFunc(ctx, agentClient, clusterID)
	if err != nil {
		return nil, err
	}

	proxy := &PremiumDMProxy{
		agentCli:    agentClient,
		DMNameSpace: fmt.Sprintf("dm%s", sharedPool.Id),
	}

	resolvedDMCluster := dmCluster
	if resolvedDMCluster == nil {
		resolvedDMCluster, err = resolveDMClusterForTiDBIDFunc(ctx, tiDBID, mgmtClient, agentCfg)
		if err != nil {
			return nil, errors.Annotate(err, "resolve dm cluster for routing")
		}
	}

	if !regionalapi.IsOperatorV2Backend(resolvedDMCluster.Backend) {
		return proxy, nil
	}
	if resolvedDMCluster.Id == "" {
		return nil, errors.New("dm cluster id is required for operator-v2 service-proxy routing")
	}
	if sharedPool.GatewayBaseUrl == "" {
		return nil, errors.New("gateway base url is required for operator-v2 service-proxy routing")
	}
	if sharedPool.Provider == nil || sharedPool.Provider.AuthToken == "" {
		return nil, errors.New("provider auth token is required for operator-v2 service-proxy routing")
	}

	dmClient, err := newDMOpenAPIClientFunc(regionalapi.DMConfig{
		GatewayBaseURL: sharedPool.GatewayBaseUrl,
		AuthToken:      sharedPool.Provider.AuthToken,
		SharedPoolID:   sharedPool.Id,
		DMClusterID:    resolvedDMCluster.Id,
	})
	if err != nil {
		return nil, err
	}

	proxy.useServiceProxy = true
	proxy.dmOpenAPI = dmClient
	proxy.dmClusterID = resolvedDMCluster.Id
	proxy.gatewayBaseURL = sharedPool.GatewayBaseUrl
	return proxy, nil
}

// ResolveDMClusterForTiDBID returns the single DM cluster for the task's shared pool.
func ResolveDMClusterForTiDBID(
	ctx context.Context,
	tiDBID string,
	mgmtClient grpcli.MgmtSvcGrpcClient,
	agentCfg agent.ClientConfig,
) (*agent.DMCluster, error) {
	clusterID, err := strconv.ParseUint(tiDBID, 10, 64)
	if err != nil {
		return nil, err
	}
	agentClient, err := newNGRegionalClientByClusterIDFunc(ctx, clusterID, mgmtClient, &agentCfg)
	if err != nil {
		return nil, err
	}
	sharedPool, err := getSharedPoolByClusterIDFunc(ctx, agentClient, clusterID)
	if err != nil {
		return nil, err
	}
	sharedPoolID, err := strconv.ParseUint(sharedPool.Id, 10, 64)
	if err != nil {
		return nil, err
	}
	listResp, err := agentClient.ListDMClusters(ctx, sharedPoolID)
	if err != nil {
		return nil, err
	}
	if len(listResp.DMClusters) == 0 {
		return nil, agent.DMClusterNotFoundErr
	}
	if len(listResp.DMClusters) > 1 {
		return nil, fmt.Errorf("shared pool %d has %d dm clusters", sharedPoolID, len(listResp.DMClusters))
	}
	return listResp.DMClusters[0], nil
}

type DMOpenapiError struct {
	ErrorCode int    `json:"error_code"`
	ErrorMsg  string `json:"error_msg"`
}

func (e *DMOpenapiError) Error() string {
	return fmt.Sprintf("DMOpenapiError(code=%d, msg=%s)", e.ErrorCode, e.ErrorMsg)
}

func handleErrorResponse(status int64, body []byte) error {
	dmOpenAPiError := &DMOpenapiError{}
	if err := json.Unmarshal(body, dmOpenAPiError); err == nil && dmOpenAPiError.ErrorCode != 0 {
		return dmOpenAPiError
	}
	return fmt.Errorf("unknown error: status=%d, resp='%s'", status, string(body))
}

func (p *PremiumDMProxy) dmOpenAPIRouteFields(method, apiPath string) []zap.Field {
	fields := []zap.Field{
		zap.String("method", method),
		zap.String("path", apiPath),
	}
	if p.useServiceProxy {
		sharedPoolID := strings.TrimPrefix(p.DMNameSpace, "dm")
		proxyPath := regionalapi.BuildDMApiPath(sharedPoolID, p.dmClusterID, apiPath)
		fields = append(fields,
			zap.String("proxy_path", proxyPath),
			zap.String("uri", regionalapi.BuildDMApiURL(p.gatewayBaseURL, sharedPoolID, p.dmClusterID, apiPath)),
			zap.String("gateway_host", gatewayHostFromBaseURL(p.gatewayBaseURL)),
			zap.String("dm_cluster_id", p.dmClusterID),
			zap.String("shared_pool_id", sharedPoolID),
		)
	} else {
		fields = append(fields, zap.String("dm_namespace", p.DMNameSpace))
	}
	return fields
}

func gatewayHostFromBaseURL(baseURL string) string {
	if baseURL == "" {
		return ""
	}
	parsed, err := url.Parse(baseURL)
	if err != nil {
		return ""
	}
	return parsed.Host
}

func redactedJSONBodyField(body []byte) zap.Field {
	if len(body) == 0 {
		return zap.Skip()
	}
	var payload any
	if err := json.Unmarshal(body, &payload); err != nil {
		return zap.ByteString("body", body)
	}
	redacted, err := redact.RedactKeysToJSON(redact.PasswordFields, payload)
	if err != nil {
		return zap.ByteString("body", body)
	}
	return zap.ByteString("body", redacted)
}

func (p *PremiumDMProxy) dmOpenAPIRequest(
	ctx context.Context, method, path string, body []byte, opts dmRequestOpts,
) (int64, []byte, error) {
	reqBody := body
	if opts.patchTargetConn && p.useServiceProxy {
		patched, err := p.patchDMTargetConn(ctx, body, opts.clusterID)
		if err != nil {
			return 0, nil, err
		}
		reqBody = patched
	}

	if p.useServiceProxy {
		status, respBody, err := p.dmOpenAPI.Do(ctx, method, path, reqBody)
		if err != nil {
			return 0, nil, err
		}
		return int64(status), respBody, nil
	}

	res, err := p.agentCli.DMOpenAPIProxy(ctx, &agentpb.DMOpenAPIProxyRequest{
		DmNamespace:     p.DMNameSpace,
		Path:            path,
		Method:          method,
		Body:            body,
		PatchTargetConn: opts.patchTargetConn,
		ClusterId:       opts.clusterID,
	})
	if err != nil {
		return 0, nil, err
	}
	return res.Status, res.Body, nil
}

func (p *PremiumDMProxy) patchDMTargetConn(ctx context.Context, body []byte, clusterID string) ([]byte, error) {
	if clusterID == "" {
		return body, nil
	}
	host, err := getTiDBHostFromAgentFunc(ctx, p.agentCli, clusterID)
	if err != nil {
		return nil, err
	}

	var payload map[string]any
	if err := json.Unmarshal(body, &payload); err != nil {
		return nil, err
	}
	task, ok := payload["task"].(map[string]any)
	if !ok {
		return body, nil
	}
	targetConfig, ok := task["target_config"].(map[string]any)
	if !ok {
		targetConfig = make(map[string]any)
		task["target_config"] = targetConfig
	}
	targetConfig["host"] = host
	return json.Marshal(payload)
}

func getTiDBHostFromAgent(ctx context.Context, agentCli *agent.NGRegionalClient, clusterID string) (string, error) {
	resp, err := agentCli.GetTiDBHost(ctx, &agentpb.GetTiDBHostRequest{ClusterId: clusterID})
	if err != nil {
		return "", err
	}
	if resp.GetHost() == "" {
		return "", errors.New("tidb host is empty")
	}
	return resp.GetHost(), nil
}

var getTiDBHostFromAgentFunc = getTiDBHostFromAgent

func (p *PremiumDMProxy) CreatePremiumDMSource(ctx context.Context, req *dm.CreateSourceRequest) (*dm.Source, error) {
	logger := pkglog.GetLogger(ctx)
	body, err := json.Marshal(req)
	if err != nil {
		return nil, err
	}
	routeFields := p.dmOpenAPIRouteFields(http.MethodPost, PremiumDMSourcePath)
	logger.Info("create premium dm source", append(routeFields, redactedJSONBodyField(body))...)
	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodPost, PremiumDMSourcePath, body, dmRequestOpts{})
	if err != nil {
		return nil, err
	}
	logger.Info("create premium dm source result",
		append(routeFields, zap.Int64("code", status), zap.ByteString("body", respBody))...)
	if status >= http.StatusBadRequest {
		return nil, handleErrorResponse(status, respBody)
	}
	var ret dm.Source
	if err := json.Unmarshal(respBody, &ret); err != nil {
		return nil, err
	}
	return &ret, nil
}

func (p *PremiumDMProxy) GetPremiumDMSource(ctx context.Context, name string) (*dm.Source, error) {
	path := fmt.Sprintf("%s/%s?with_status=true", PremiumDMSourcePath, name)
	routeFields := p.dmOpenAPIRouteFields(http.MethodGet, path)
	logger := pkglog.GetLogger(ctx).With(append(routeFields, zap.String("source", name))...)
	logger.Info("get premium dm source")

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodGet, path, nil, dmRequestOpts{})
	if err != nil {
		return nil, err
	}

	logger.Info("get premium dm source result", zap.ByteString("body", respBody), zap.Int64("code", status))
	if status >= http.StatusBadRequest {
		return nil, handleErrorResponse(status, respBody)
	}

	var ret dm.Source
	if err := json.Unmarshal(respBody, &ret); err != nil {
		return nil, err
	}
	return &ret, nil
}

func (p *PremiumDMProxy) DeletePremiumDMSource(ctx context.Context, name string) error {
	path := fmt.Sprintf("%s/%s", PremiumDMSourcePath, name)
	routeFields := p.dmOpenAPIRouteFields(http.MethodDelete, path)
	logger := pkglog.GetLogger(ctx).With(append(routeFields, zap.String("source", name))...)
	logger.Info("delete premium dm source")

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodDelete, path, nil, dmRequestOpts{})
	if err != nil {
		return err
	}

	logger.Info("delete premium dm source result", zap.Int64("code", status), zap.String("body", string(respBody)))
	if status >= http.StatusBadRequest {
		return handleErrorResponse(status, respBody)
	}
	return nil
}

func (p *PremiumDMProxy) CreatePremiumDMTask(
	ctx context.Context, req *dm.CreateTaskRequest, clusterID string,
) (*dm.OperateTaskResponse, error) {
	logger := pkglog.GetLogger(ctx)
	body, err := json.Marshal(req)
	if err != nil {
		return nil, err
	}
	routeFields := p.dmOpenAPIRouteFields(http.MethodPost, PremiumDMTaskPath)
	logger.Info("create premium dm task", append(routeFields, redactedJSONBodyField(body))...)

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodPost, PremiumDMTaskPath, body, dmRequestOpts{
		patchTargetConn: true,
		clusterID:       clusterID,
	})
	if err != nil {
		return nil, err
	}

	logger.Info("create premium dm task result",
		append(routeFields, zap.Int64("code", status), zap.ByteString("body", respBody))...)
	if status >= http.StatusBadRequest {
		return nil, handleErrorResponse(status, respBody)
	}

	var ret dm.OperateTaskResponse
	if err := json.Unmarshal(respBody, &ret); err != nil {
		return nil, err
	}
	return &ret, nil
}

func (p *PremiumDMProxy) DeletePremiumDMTask(ctx context.Context, name string) error {
	path := fmt.Sprintf("%s/%s?force=true", PremiumDMTaskPath, name)
	routeFields := p.dmOpenAPIRouteFields(http.MethodDelete, path)
	logger := pkglog.GetLogger(ctx).With(append(routeFields, zap.String("task", name))...)
	logger.Info("delete premium dm task")

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodDelete, path, nil, dmRequestOpts{})
	if err != nil {
		return err
	}

	logger.Info("delete premium dm task result", zap.Int64("code", status), zap.String("body", string(respBody)))
	if status >= http.StatusBadRequest {
		return handleErrorResponse(status, respBody)
	}
	return nil
}

func (p *PremiumDMProxy) doTaskAction(
	ctx context.Context,
	name string,
	action string,
	req any,
) error {
	path := fmt.Sprintf("%s/%s/%s", PremiumDMTaskPath, name, action)
	body, _ := json.Marshal(req)
	routeFields := p.dmOpenAPIRouteFields(http.MethodPost, path)
	logger := pkglog.GetLogger(ctx).With(append(routeFields, zap.String("task", name))...)
	logger.Info(action+" premium dm task", redactedJSONBodyField(body))

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodPost, path, body, dmRequestOpts{})
	if err != nil {
		return err
	}

	logger.Info(action+" premium dm task result", zap.Int64("code", status), zap.String("body", string(respBody)))
	if status >= http.StatusBadRequest {
		return handleErrorResponse(status, respBody)
	}
	return nil
}

func (p *PremiumDMProxy) StartPremiumDMTask(ctx context.Context, name string, req *dm.StartTaskRequest) error {
	return p.doTaskAction(ctx, name, "start", req)
}

func (p *PremiumDMProxy) StopPremiumDMTask(ctx context.Context, name string, req *dm.StopTaskRequest) error {
	return p.doTaskAction(ctx, name, "stop", req)
}

func (p *PremiumDMProxy) QueryPremiumDMTaskStatus(
	ctx context.Context, name string, sourceNameList []string,
) (*dm.GetTaskStatusResponse, error) {
	path := fmt.Sprintf("%s/%s/status?source_name_list=%s", PremiumDMTaskPath, name, sourceNameList)
	routeFields := p.dmOpenAPIRouteFields(http.MethodGet, path)
	logger := pkglog.GetLogger(ctx).With(append(routeFields, zap.String("task", name))...)
	logger.Info("query premium dm task status")

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodGet, path, nil, dmRequestOpts{})
	if err != nil {
		return nil, err
	}

	logger.Info("query premium dm task status result", zap.Int64("code", status), zap.String("body", string(respBody)))
	if status >= http.StatusBadRequest {
		return nil, handleErrorResponse(status, respBody)
	}

	var ret dm.GetTaskStatusResponse
	if err := json.Unmarshal(respBody, &ret); err != nil {
		return nil, err
	}
	return &ret, nil
}

func (p *PremiumDMProxy) ListPremiumDMWorkers(ctx context.Context) (*dm.GetClusterWorkerListResponse, error) {
	routeFields := p.dmOpenAPIRouteFields(http.MethodGet, PremiumDMWorkerPath)
	logger := pkglog.GetLogger(ctx).With(routeFields...)
	logger.Info("list premium dm workers")

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodGet, PremiumDMWorkerPath, nil, dmRequestOpts{})
	if err != nil {
		return nil, err
	}

	logger.Info("list premium dm workers result", zap.Int64("code", status), zap.String("body", string(respBody)))
	if status >= http.StatusBadRequest {
		return nil, handleErrorResponse(status, respBody)
	}

	var ret dm.GetClusterWorkerListResponse
	if err := json.Unmarshal(respBody, &ret); err != nil {
		return nil, err
	}
	return &ret, nil
}

func (p *PremiumDMProxy) DeletePremiumDMWorkers(ctx context.Context, workerName string) error {
	path := fmt.Sprintf("%s/%s", PremiumDMWorkerPath, workerName)
	routeFields := p.dmOpenAPIRouteFields(http.MethodDelete, path)
	logger := pkglog.GetLogger(ctx).With(routeFields...)
	logger.Info("delete premium dm workers")

	status, respBody, err := p.dmOpenAPIRequest(ctx, http.MethodDelete, path, nil, dmRequestOpts{})
	if err != nil {
		return err
	}

	logger.Info("list premium dm workers result", zap.Int64("code", status), zap.String("body", string(respBody)))
	if status >= http.StatusBadRequest {
		return handleErrorResponse(status, respBody)
	}
	return nil
}