package agent

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

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/tidbcloud/dataflow-service-ng/pkg/redact"
	agentpb "github.com/tidbcloud/dataflow-service/dataflow-agent-ng/idl/pbgen/proto/ng_agent"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
)

func (c *NGRegionalClient) GetPublicKey(ctx context.Context, req *agentpb.GetPublicKeyRequest) (*agentpb.GetPublicKeyResponse, error) {
	url := fmt.Sprintf("%s/api/v1/crypto/pem", c.AgentUrl)

	respBytes, err := c.getResponseProtoJson(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agentpb.GetPublicKeyResponse
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (c *NGRegionalClient) GetTiDBHost(ctx context.Context, req *agentpb.GetTiDBHostRequest) (*agentpb.GetTiDBHostResponse, error) {
	url := fmt.Sprintf("%s/api/v1/tidb/host", c.AgentUrl)

	respBytes, err := c.getResponseProtoJson(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agentpb.GetTiDBHostResponse
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (c *NGRegionalClient) ExecSQLOnDMSource(ctx context.Context, req *agentpb.ExecSQLOnDMSourceRequest) (
	*agentpb.ExecSQLOnDMSourceResponse, error,
) {
	url := fmt.Sprintf("%s/api/v1/dm/source/exec_sql", c.AgentUrl)

	respBytes, err := c.getResponseProtoJson(ctx, http.MethodPost, url, req)
	if err != nil {
		if msg := dmSourceProxyPayloadErrorMessage(err); msg != "" {
			return &agentpb.ExecSQLOnDMSourceResponse{ErrorMsg: msg}, nil
		}
		return nil, err
	}
	var resp agentpb.ExecSQLOnDMSourceResponse
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func dmSourceProxyPayloadErrorMessage(err error) string {
	var upstreamErr *retryableUpstreamError
	if !stderrors.As(err, &upstreamErr) || upstreamErr.statusCode != http.StatusGatewayTimeout {
		return ""
	}

	var errResp ErrorResponse
	if parseErr := json.Unmarshal([]byte(upstreamErr.response), &errResp); parseErr != nil {
		return ""
	}
	if !isDMSourceDBNetworkErrorMessage(errResp.Message) {
		return ""
	}
	return errResp.Message
}

func isDMSourceDBNetworkErrorMessage(msg string) bool {
	msg = strings.ToLower(msg)
	return strings.Contains(msg, "dial tcp") ||
		strings.Contains(msg, "i/o timeout") ||
		strings.Contains(msg, "connection refused") ||
		strings.Contains(msg, "connection reset") ||
		strings.Contains(msg, "no such host") ||
		strings.Contains(msg, "network is unreachable")
}

func (c *NGRegionalClient) DMOpenAPIProxy(ctx context.Context, req *agentpb.DMOpenAPIProxyRequest) (*agentpb.DMOpenAPIProxyResponse, error) {
	url := fmt.Sprintf("%s/api/v1/dm/openapi/proxy", c.AgentUrl)

	respBytes, err := c.getResponseProtoJson(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agentpb.DMOpenAPIProxyResponse
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	log.Info("get dm response", zap.String("response", string(respBytes)))
	return &resp, nil
}

func (c *NGRegionalClient) PrecheckDMJob(ctx context.Context, req *agentpb.PrecheckDMJobReq) (*agentpb.PrecheckDMJobResp, error) {
	url := fmt.Sprintf("%s/api/v1/dm/jobprecheck", c.AgentUrl)
	respBytes, err := c.getResponseProtoJsonWithRetry(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agentpb.PrecheckDMJobResp
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (c *NGRegionalClient) GetDMClusterByID(ctx context.Context, sharedPoolId uint64, dmClusterID string) (*DMCluster, error) {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d/dmClusters/%s", c.BaseUrl, sharedPoolId, dmClusterID)
	respBytes, err := c.getResponseWithRetry(ctx, "GET", url, nil)
	if err != nil {
		// check if dm cluster not found
		var errResp *ErrorResponse
		if stderrors.As(err, &errResp) && errResp.Code == 5 { // grpc not found code
			log.Info("dm cluster not found", zap.String("dmClusterID", dmClusterID))
			return nil, DMClusterNotFoundErr
		}
		log.Error("failed to get dm cluster", zap.String("dmClusterID", dmClusterID), zap.Error(err))
		return nil, errors.Trace(err)
	}
	var dmCluster DMCluster
	if err := json.Unmarshal(respBytes, &dmCluster); err != nil {
		log.Error("unmarshal dm cluster failed", zap.String("dmClusterID", dmClusterID), zap.String("resp", string(respBytes)), zap.Error(err))
		return nil, errors.Trace(err)
	}
	log.Info("success to get dm cluster", zap.String("dmClusterID", dmClusterID), zap.String("respBytes", string(respBytes)))
	return &dmCluster, nil
}

func (c *NGRegionalClient) ListDMClusters(ctx context.Context, sharedPoolId uint64) (*ListDMClustersResponse, error) {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d/dmClusters", c.BaseUrl, sharedPoolId)
	respBytes, err := c.getResponseWithRetry(ctx, http.MethodGet, url, nil)
	if err != nil {
		log.Error("failed to list dm clusters", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("respBytes", string(respBytes)), zap.Error(err))
		return nil, errors.Trace(err)
	}
	var dmClustersResp ListDMClustersResponse
	if err := json.Unmarshal(respBytes, &dmClustersResp); err != nil {
		log.Error("unmarshal dm clusters failed", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("resp", string(respBytes)), zap.Error(err))
		return nil, errors.Trace(err)
	}
	log.Info("success to list dm clusters", zap.Uint64("sharedPoolId", sharedPoolId), zap.Int("dmClusterCount", len(dmClustersResp.DMClusters)),
		zap.Int32("totalCount", dmClustersResp.TotalCount))
	return &dmClustersResp, nil
}

func (c *NGRegionalClient) ScaleDMCluster(ctx context.Context, sharedPoolId uint64, dmClusterID string, req *ScaleDMClusterRequest) error {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d/dmClusters/%s:scale", c.BaseUrl, sharedPoolId, dmClusterID)
	payload, err := json.Marshal(req)
	if err != nil {
		return errors.Trace(err)
	}
	respBytes, err := c.getResponseWithRetry(ctx, "POST", url, payload)
	if err != nil {
		log.Error("failed to scale dm cluster", zap.String("dmClusterID", dmClusterID), zap.String("payload", string(payload)), zap.String("respBytes", string(respBytes)), zap.Error(err))
		return err
	}
	log.Info("scale dm cluster", zap.String("dmClusterID", dmClusterID), zap.String("payload", string(payload)), zap.String("respBytes", string(respBytes)))
	return nil
}

func (c *NGRegionalClient) DeleteDMCluster(ctx context.Context, sharedPoolId uint64, dmClusterID string) error {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d/dmClusters/%s", c.BaseUrl, sharedPoolId, dmClusterID)
	respBytes, err := c.getResponseWithRetry(ctx, http.MethodDelete, url, nil)
	if err != nil {
		log.Error("failed to delete dm cluster", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("dmClusterID", dmClusterID), zap.String("respBytes", string(respBytes)), zap.Error(err))
		return err
	}
	log.Info("delete dm cluster", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("dmClusterID", dmClusterID), zap.String("respBytes", string(respBytes)))
	return nil
}

func (c *NGRegionalClient) CreateDMCluster(ctx context.Context, sharedPoolId uint64, req *CreateDMClusterRequest) (*DMCluster, error) {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d/dmClusters", c.BaseUrl, sharedPoolId)
	payload, err := json.Marshal(req)
	if err != nil {
		return nil, errors.Trace(err)
	}
	respBytes, err := c.getResponseWithRetry(ctx, "POST", url, payload)
	if err != nil {
		log.Error("failed to create dm cluster", zap.String("payload", string(payload)), zap.String("respBytes", string(respBytes)), zap.Error(err))
		return nil, err
	}
	var dmCluster DMCluster
	if err := json.Unmarshal(respBytes, &dmCluster); err != nil {
		log.Error("unmarshal dm cluster failed", zap.String("resp", string(respBytes)), zap.Error(err))
		return nil, errors.Trace(err)
	}
	log.Info("success to create dm cluster", zap.String("respBytes", string(respBytes)))
	return &dmCluster, nil
}

func (c *NGRegionalClient) ExecuteSQL(ctx context.Context, req *agentpb.ExecSQLReq) (*agentpb.ExecSQLResp, error) {
	url := fmt.Sprintf("%s/api/v1/cluster/execsql", c.AgentUrl)
	respBytes, err := c.getResponseProtoJson(ctx, http.MethodPost, url, req)
	if err != nil {
		log.Error("execute sql failed", zap.Error(err))
		return nil, err
	}
	var resp agentpb.ExecSQLResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (c *NGRegionalClient) GetTiDBGroup(ctx context.Context, clusterID uint64) (*TiDBGroup, error) {
	url := fmt.Sprintf("%s/api/premium/v1/clusters/%d/tidbGroups?clusterId=%d", c.BaseUrl, clusterID, clusterID)
	respBytes, err := c.getResponseWithRetry(ctx, "GET", url, nil)
	if err != nil {
		log.Error("failed to get tidb group", zap.String("resp", string(respBytes)), zap.Error(err))
		return nil, err
	}
	var groupsResp struct {
		TidbGroups []*TiDBGroup `json:"tidbGroups"`
		TotalCount uint64       `json:"totalCount"`
	}
	if err := json.Unmarshal(respBytes, &groupsResp); err != nil {
		log.Error("unmarshal tidb group failed", zap.String("resp", string(respBytes)), zap.Error(err))
		return nil, errors.Trace(err)
	}
	if groupsResp.TotalCount == 0 {
		log.Error("get tidb group failed", zap.String("resp", string(respBytes)), zap.Error(err))
		return nil, errors.Trace(err)
	}
	log.Info("Get TiDB Group", zap.Uint64("clusterID", clusterID), zap.Any("group", groupsResp.TidbGroups[0]))
	return groupsResp.TidbGroups[0], nil
}

func (c *NGRegionalClient) GetImportRoleConfig(ctx context.Context, clusterID uint64) (*ImportRoleConfig, error) {
	url := fmt.Sprintf("%s/api/premium/v1/clusters/%d/importRoleConfig", c.BaseUrl, clusterID)
	respBytes, err := c.getResponseWithRetry(ctx, http.MethodGet, url, nil)
	if err != nil {
		log.Error("failed to get import role config", zap.Uint64("clusterID", clusterID), zap.Error(err))
		return nil, err
	}
	var resp ImportRoleConfig
	if err := json.Unmarshal(respBytes, &resp); err != nil {
		log.Error("unmarshal import role config failed", zap.Uint64("clusterID", clusterID), zap.Error(err))
		return nil, errors.Trace(err)
	}
	resp.RoleArn = strings.TrimSpace(resp.RoleArn)
	resp.ExternalID = strings.TrimSpace(resp.ExternalID)
	if resp.RoleArn == "" {
		return nil, errors.Errorf("import role ARN is empty for cluster %d", clusterID)
	}
	if resp.ExternalID == "" {
		return nil, errors.Errorf("import role external ID is empty for cluster %d", clusterID)
	}
	return &resp, nil
}

func ApplyImportRoleConfigToStorageURI(storageURI string, roleConfig *ImportRoleConfig) (string, error) {
	parsedURI, err := url.Parse(storageURI)
	if err != nil {
		return "", err
	}
	if !strings.EqualFold(parsedURI.Scheme, "s3") && !strings.EqualFold(parsedURI.Scheme, "oss") {
		return storageURI, nil
	}
	if roleConfig == nil {
		return "", errors.New("import role config is nil")
	}
	roleArn := strings.TrimSpace(roleConfig.RoleArn)
	if roleArn == "" {
		return "", errors.New("import role ARN is empty")
	}
	query := parsedURI.Query()
	query.Del("access-key")
	query.Del("secret-access-key")
	query.Del("session-token")
	query.Del("external-id")
	query.Set("role-arn", roleArn)
	if externalID := strings.TrimSpace(roleConfig.ExternalID); externalID != "" {
		query.Set("external-id", externalID)
	}
	parsedURI.RawQuery = query.Encode()
	return parsedURI.String(), nil
}

func (c *NGRegionalClient) ScaleSharedPool(ctx context.Context, sharedPoolId uint64, req *ScaleSharedPoolRequest) error {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d:scale", c.BaseUrl, sharedPoolId)
	payload, err := json.Marshal(req)
	if err != nil {
		return errors.Trace(err)
	}
	respBytes, err := c.getResponseWithRetry(ctx, "POST", url, payload)
	if err != nil {
		log.Error("failed to scale tidb group dxf", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("payload", string(payload)), zap.String("respBytes", string(respBytes)), zap.Error(err))
		return err
	}
	log.Info("scale tidb group dxf", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("payload", string(payload)), zap.String("respBytes", string(respBytes)))
	return nil
}

var sharedPoolLogRedactKeys = []string{"auth_token", "authtoken"}

func redactSharedPoolResponseForLog(respBytes []byte) string {
	if len(respBytes) == 0 {
		return ""
	}
	var payload any
	if err := json.Unmarshal(respBytes, &payload); err != nil {
		return string(respBytes)
	}
	redacted, err := redact.AsMapRedactKeys(sharedPoolLogRedactKeys, payload)
	if err != nil {
		return string(respBytes)
	}
	b, err := json.Marshal(redacted)
	if err != nil {
		return string(respBytes)
	}
	return string(b)
}

func (c *NGRegionalClient) GetSharePool(ctx context.Context, sharedPoolId uint64) (*SharedPool, error) {
	url := fmt.Sprintf("%s/api/premium/v1/sharedPools/%d", c.BaseUrl, sharedPoolId)
	respBytes, err := c.getResponseWithRetry(ctx, "GET", url, nil)
	logResp := redactSharedPoolResponseForLog(respBytes)
	if err != nil {
		log.Error("failed to get share pool", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("respBytes", logResp), zap.Error(err))
		return nil, errors.Trace(err)
	}
	var sharedPools SharedPool
	if err := json.Unmarshal(respBytes, &sharedPools); err != nil {
		log.Error("unmarshal share pool failed", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("resp", logResp), zap.Error(err))
		return nil, errors.Trace(err)
	}
	log.Info("success to get share pool", zap.Uint64("sharedPoolId", sharedPoolId), zap.String("respBytes", logResp))
	return &sharedPools, nil
}

func (c *NGRegionalClient) GetSharePoolByClusterID(ctx context.Context, clusterID uint64) (*SharedPool, error) {
	tidbGroup, err := c.GetTiDBGroup(ctx, clusterID)
	if err != nil {
		log.Error("failed to get tidb group", zap.Uint64("clusterID", clusterID), zap.Error(err))
		return nil, errors.Trace(err)
	}

	sharedPoolId, err := strconv.ParseUint(tidbGroup.SharedPoolId, 10, 64)
	if err != nil {
		log.Error("failed to parse shared pool id", zap.Uint64("clusterID", clusterID), zap.String("sharedPoolId", tidbGroup.SharedPoolId), zap.Error(err))
		return nil, errors.Trace(err)
	}

	sharedPool, err := c.GetSharePool(ctx, sharedPoolId)
	if err != nil {
		log.Error("failed to get shared pool", zap.Uint64("clusterID", clusterID), zap.Uint64("sharedPoolId", sharedPoolId), zap.Error(err))
		return nil, errors.Trace(err)
	}
	return sharedPool, nil
}