package agent

/*
This package is ported from github.com/tidbcloud/dataflow-service/dataflow-service-ng/pkg/agent/dm, with some adaptations.
*/

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	stderrors "errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/avast/retry-go/v4"
	"github.com/gin-gonic/gin/binding"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	pkglog "github.com/tidbcloud/dataflow-service-ng/pkg/log"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

// ErrorResponse represents an error response from the agent.
type ErrorResponse struct {
	Code    int64       `json:"code"`
	Message string      `json:"message"`
	Details interface{} `json:"details"`
}

func (err *ErrorResponse) Error() string {
	return err.Message
}

var errRetryableUpstream = stderrors.New("retryable upstream error")

type retryableUpstreamError struct {
	statusCode int
	response   string
	err        error
}

func (e *retryableUpstreamError) Error() string {
	if e == nil {
		return ""
	}
	if e.err != nil {
		return e.err.Error()
	}
	return "retryable upstream error"
}

func (e *retryableUpstreamError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.err
}

func (e *retryableUpstreamError) Is(target error) bool {
	return target == errRetryableUpstream
}

func isRetryableUpstreamError(err error) bool {
	if err == nil {
		return false
	}
	if stderrors.Is(err, errRetryableUpstream) {
		return true
	}

	var urlErr *url.Error
	if stderrors.As(err, &urlErr) {
		if urlErr.Timeout() {
			return true
		}
		err = urlErr.Err
	}

	var netErr net.Error
	if stderrors.As(err, &netErr) && (netErr.Timeout()) {
		return true
	}

	errMsg := strings.ToLower(err.Error())
	return strings.Contains(errMsg, "connection reset") ||
		strings.Contains(errMsg, "connection refused") ||
		strings.Contains(errMsg, "server closed idle connection") ||
		strings.Contains(errMsg, "unexpected eof") ||
		strings.Contains(errMsg, "eof")
}

// ClientConfig holds the configuration needed to create NGRegionalClient instances.
type ClientConfig struct {
	MgmtSvcProxyAddr  string
	MgmtSvcProxyToken string
}

type NGRegionalClient struct {
	BaseUrl  string
	AgentUrl string
	token    string
}

var PEMChangedError = errors.New("pem changed")

var (
	DMClusterNotFoundErr = errors.New("dm cluster not found")
)

type ngRegionalClientProvider struct {
	mgmtClient grpcli.MgmtSvcGrpcClient
	cfg        *ClientConfig
}
type NGRegionalClientProvider interface {
	NewByClusterID(ctx context.Context, clusterID uint64) (*NGRegionalClient, error)
}

func NewNGRegionalClientProvider(mgmtClient grpcli.MgmtSvcGrpcClient, cfg *ClientConfig) *ngRegionalClientProvider {
	return &ngRegionalClientProvider{
		mgmtClient: mgmtClient,
		cfg:        cfg,
	}
}

func (p *ngRegionalClientProvider) NewByClusterID(ctx context.Context, clusterID uint64) (*NGRegionalClient, error) {
	return NewNGRegionalClientByClusterID(
		ctx,
		clusterID,
		p.mgmtClient,
		&ClientConfig{
			MgmtSvcProxyAddr:  p.cfg.MgmtSvcProxyAddr,
			MgmtSvcProxyToken: p.cfg.MgmtSvcProxyToken,
		},
	)
}

func NewNGRegionalClientByClusterID(ctx context.Context, clusterID uint64, mgmtClient grpcli.MgmtSvcGrpcClient, cfg *ClientConfig) (*NGRegionalClient, error) {
	ctx, cancelFunc := context.WithTimeout(ctx, time.Second*10)
	defer cancelFunc()
	listResp, err := mgmtClient.ListClusters(ctx, &mgmt.ListClustersRequest{Filter: &mgmt.ClusterFilter{ClusterIds: []uint64{clusterID}}})
	if err != nil {
		log.Error("list mgmt clusters error", zap.Error(err), zap.Uint64("clusterID", clusterID))
		return nil, fmt.Errorf("list clusters by ID error: %w", err)
	}
	if len(listResp.Clusters) == 0 {
		return nil, fmt.Errorf("cluster not found for id %d", clusterID)
	}

	if listResp.Clusters[0].GetNextGenCluster() == nil || listResp.Clusters[0].GetNextGenCluster().GetSharedPoolId() == "" {
		return nil, errors.New("cluster is not a nextgen cluster or shared pool id is empty")
	}

	sharedPoolId, perr := strconv.ParseUint(listResp.Clusters[0].GetNextGenCluster().GetSharedPoolId(), 10, 64)
	if perr != nil {
		return nil, errors.Annotate(perr, "parse shared pool id error")
	}

	return NewNGRegionalClientByResourcePoolID(ctx, sharedPoolId, mgmtClient, cfg)
}

func NewNGRegionalClientByResourcePoolID(ctx context.Context, resourcePoolID uint64, mgmtClient grpcli.MgmtSvcGrpcClient, cfg *ClientConfig) (*NGRegionalClient, error) {
	ctx, cancelFunc := context.WithTimeout(ctx, time.Second*10)
	defer cancelFunc()

	// check resource pool existence
	_, err := mgmtClient.GetResourcePoolByID(ctx, resourcePoolID)
	if err != nil {
		log.Error("get mgmt resource pool error", zap.Error(err), zap.Uint64("resourcePoolID", resourcePoolID))
		return nil, errors.Annotate(err, "get resource pool by ID error")
	}

	baseURL := fmt.Sprintf("%s/proxy/nextgen/resourcePools/%d", cfg.MgmtSvcProxyAddr, resourcePoolID)
	client := &NGRegionalClient{
		BaseUrl: baseURL,
		token:   cfg.MgmtSvcProxyToken,
	}

	client.AgentUrl = baseURL + fmt.Sprintf("/dataplane/proxy/sharedPools/%d/dataflow-agent", resourcePoolID)
	return client, nil
}

const (
	ngRegionalRetryAttempts = 3
	ngRegionalRetryDelay    = 200 * time.Millisecond
)

var sharedHttpClient = &http.Client{
	Transport: otelhttp.NewTransport(&http.Transport{
		TLSClientConfig:     &tls.Config{InsecureSkipVerify: true}, // TODO: remove InsecureSkipVerify after the agent supports TLS with valid certs
		MaxIdleConns:        500,
		IdleConnTimeout:     90 * time.Second,
		MaxIdleConnsPerHost: 100,
	}),
}

func (c *NGRegionalClient) getResponseWithRetry(ctx context.Context, method, url string, payload []byte) ([]byte, error) {
	var respBytes []byte
	err := retry.Do(
		func() error {
			var err error
			respBytes, err = c.getResponse(ctx, method, url, payload)
			return err
		},
		retry.Attempts(ngRegionalRetryAttempts),
		retry.Delay(ngRegionalRetryDelay),
		retry.DelayType(retry.FixedDelay),
		retry.Context(ctx),
		retry.LastErrorOnly(true),
		retry.OnRetry(func(n uint, err error) {
			pkglog.GetLogger(ctx).Warn("retry ng regional request",
				zap.Uint("attempt", n+1),
				zap.String("method", method),
				zap.String("url", url),
				zap.Error(err),
			)
		}),
		retry.RetryIf(func(err error) bool {
			return isRetryableUpstreamError(err)
		}),
	)
	if err != nil {
		return nil, err
	}
	return respBytes, nil
}

func (c *NGRegionalClient) getResponse(ctx context.Context, method, url string, payload []byte) ([]byte, error) {
	logger := pkglog.GetLogger(ctx).With(zap.String("method", method), zap.String("url", url))
	httpReq, err := http.NewRequestWithContext(ctx, method, url, bytes.NewReader(payload))
	if err != nil {
		return nil, errors.Trace(err)
	}
	httpReq.Header.Set("Content-Type", binding.MIMEJSON)
	httpReq.Header.Set("Authorization", "Bearer "+c.token)

	httpClient := sharedHttpClient
	httpResp, err := httpClient.Do(httpReq)
	if err != nil {
		traceErr := errors.Trace(err)
		if isRetryableUpstreamError(traceErr) {
			return nil, &retryableUpstreamError{err: traceErr}
		}
		return nil, traceErr
	}
	defer httpResp.Body.Close() // nolint: errcheck

	respBytes, err := io.ReadAll(httpResp.Body)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if httpResp.StatusCode != http.StatusOK {
		if httpResp.StatusCode < 500 {
			var errResp = ErrorResponse{}
			if parseErr := json.Unmarshal(respBytes, &errResp); parseErr == nil {
				logger.Info("response failed but been catch", zap.String("response", string(respBytes)))
				return nil, &errResp
			}
		} else {
			if httpResp.StatusCode == http.StatusBadGateway ||
				httpResp.StatusCode == http.StatusServiceUnavailable ||
				httpResp.StatusCode == http.StatusGatewayTimeout {
				logger.Warn("retryable upstream response failed",
					zap.Int("status_code", httpResp.StatusCode),
					zap.String("response", string(respBytes)),
				)
				return nil, &retryableUpstreamError{
					statusCode: httpResp.StatusCode,
					response:   string(respBytes),
					err:        errors.New("request failed"),
				}
			}
		}
		logger.Error("response failed", zap.String("response", string(respBytes)))
		return nil, errors.New("request failed")
	}

	return respBytes, nil
}

func (c *NGRegionalClient) getResponseProtoJsonWithRetry(ctx context.Context, method, url string, req proto.Message) ([]byte, error) {
	payload, err := protojson.Marshal(req)
	if err != nil {
		return nil, err
	}
	return c.getResponseWithRetry(ctx, method, url, payload)
}

func (c *NGRegionalClient) getResponseProtoJson(ctx context.Context, method, url string, req proto.Message) ([]byte, error) {
	payload, err := protojson.Marshal(req)
	if err != nil {
		return nil, err
	}
	return c.getResponse(ctx, method, url, payload)
}