package http

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/url"
	"time"

	"github.com/go-resty/resty/v2"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/dataflow-service/pkg/redact"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

const (
	defaultRetryCount = 3
	minRetryWaitTime  = time.Millisecond * 500
	maxRetryWaitTime  = time.Second * 5
)

type ErrorTransformer func(code int, body []byte) bizerr.BizError

type RestyClient struct {
	cli *resty.Client
}

type RestyCliOpt func(cli *resty.Client)

type RestyRequestOpt func(req *resty.Request)

func RestyCliWithDefaultRetry() RestyCliOpt {
	return func(cli *resty.Client) {
		client := &RestyClient{cli: cli}
		cli.SetRetryCount(defaultRetryCount).
			SetRetryWaitTime(minRetryWaitTime).
			SetRetryMaxWaitTime(maxRetryWaitTime).
			AddRetryHook(func(response *resty.Response, err error) {
				if response != nil && response.IsSuccess() {
					return
				}

				var ctx context.Context
				if response != nil && response.Request != nil {
					ctx = response.Request.Context()
				} else {
					ctx = context.Background()
				}
				logger := log.GetLogger(ctx)
				if response != nil {
					logger = logger.With(zap.String("status", response.Status()))
					if response.Request != nil {
						logger = client.loggerWithDesensitizedRequest(logger, response.Request.Body)
						logger = logger.With(zap.String("method", response.Request.Method),
							zap.String("url", redact.URI(response.Request.URL)),
							zap.Any("http_tracing", response.Request.TraceInfo()))
					}
				}
				if err != nil {
					logger = logger.With(zap.String("error", redact.Text(err.Error())))
				}
				logger.Warn("http request with retry error")
			}).
			AddRetryCondition(func(response *resty.Response, err error) bool {
				if err != nil {
					return true
				}
				return response.StatusCode() >= http.StatusInternalServerError
			})
	}
}

func RestyWithBaseURL(baseURL string) RestyCliOpt {
	return func(cli *resty.Client) {
		cli.SetBaseURL(baseURL)
	}
}

func marshallerWrapper(v interface{}) ([]byte, error) {
	m, ok := v.(proto.Message)
	if ok {
		return protojson.Marshal(m)
	} else {
		return json.Marshal(v)
	}
}

func unmarshallerWrapper(data []byte, v interface{}) error {
	m, ok := v.(proto.Message)
	if ok {
		return protojson.Unmarshal(data, m)

	} else {
		return json.Unmarshal(data, v)
	}
}

func RestyWithProtoJson() RestyCliOpt {
	return func(cli *resty.Client) {
		cli.SetJSONMarshaler(marshallerWrapper).
			SetJSONUnmarshaler(unmarshallerWrapper)
	}
}

func NewRestyCli(opts ...RestyCliOpt) *RestyClient {
	// This wrapper owns request/error logging. Resty's default logger writes
	// unsanitized transport URLs (and, in debug mode, payloads) to stderr.
	cli := resty.New().SetLogger(silentRestyLogger{})
	// enable trace propagation
	cli = cli.SetTransport(otelhttp.NewTransport(cli.GetClient().Transport))
	// enable http tracing, refer to https://go.dev/blog/http-tracing
	cli = cli.EnableTrace()
	// set default timeout 10 minutes, scanning tables may spend much time for ticdc new arch
	cli = cli.SetTimeout(time.Second * 60 * 10)

	for _, opt := range opts {
		opt(cli)
	}

	return &RestyClient{
		cli: cli,
	}
}

type silentRestyLogger struct{}

func (silentRestyLogger) Errorf(string, ...interface{}) {}
func (silentRestyLogger) Warnf(string, ...interface{})  {}
func (silentRestyLogger) Debugf(string, ...interface{}) {}

func (c *RestyClient) Call(
	ctx context.Context,
	method, url string,
	pathParams map[string]string,
	queryParams url.Values,
	body interface{},
	transformer ErrorTransformer,
	opts ...RestyRequestOpt,
) ([]byte, bizerr.BizError) {
	req := c.cli.R().
		SetContext(ctx).
		SetPathParams(pathParams).
		SetQueryParamsFromValues(queryParams).
		SetBody(body)
	for _, opt := range opts {
		opt(req)
	}

	resp, err := req.Execute(method, url)
	logger := log.GetLogger(ctx).With(
		zap.String("method", method),
		zap.String("url", redact.URI(url)),
		zap.Any("http_trace", req.TraceInfo()))
	// Sanitize a separate representation; never mutate the connection inputs.
	logger = c.loggerWithDesensitizedRequest(logger, body)
	if resp != nil {
		logger = logger.With(
			zap.Duration("request_duration", resp.Time()),
			zap.String("response_status", resp.Status()))
	}

	if err != nil {
		if bizErr, ok := err.(bizerr.BizError); ok {
			err = bizerr.Restore(bizErr.ErrCode(), redact.Text(bizErr.ErrMsg()))
		} else {
			err = errors.New(redact.Text(err.Error()))
		}
		logger.Error("http request error", zap.Error(err))
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "execute request error")
	}
	if !resp.IsSuccess() {
		logger.Warn("http request not success")
		return nil, transformer(resp.StatusCode(), resp.Body())
	}

	logger.Info("http request OK")
	return resp.Body(), nil
}

// loggerWithDesensitizedRequest preserves useful request diagnostics while
// masking both secret fields and credentials embedded in string values.
func (c *RestyClient) loggerWithDesensitizedRequest(logger *zap.Logger, body interface{}) *zap.Logger {
	if body == nil {
		return logger
	}
	var payload []byte
	var err error
	switch value := body.(type) {
	case string:
		payload = []byte(value)
	case []byte:
		// json.Marshal would base64-encode raw bytes, hiding secrets from redaction.
		payload = value
	default:
		payload, err = marshallerWrapper(body)
	}
	if err != nil || !json.Valid(payload) {
		// Do not print serialization errors: custom marshalers may include secrets.
		// Opaque or malformed bodies cannot be safely inspected for secret fields.
		return logger.With(zap.String("request_body", "[REDACTED]"))
	}
	return logger.With(zap.String("request_body", redact.Text(string(payload))))
}