package service

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode"

	clinic_internal_api "github.com/pingcap-inc/clinic/clinic_data_service/pkg/clinic-internal-api"
	clinictracing "github.com/pingcap-inc/clinic/clinic_data_service/pkg/tracing"
)

const (
	byocContinuousProfilingGroupProfilesPath = "/diagnosis/api/v1/continuous_profiling/group_profiles"
	byocContinuousProfilingGroupDetailPath   = "/diagnosis/api/v1/continuous_profiling/group_profile/detail"
	maxBYOCContinuousProfilingResponse       = 16 << 20
	byocContinuousProfilingTimeout           = 180 * time.Second
	maxBYOCContinuousProfilingRangeSeconds   = 3 * 60 * 60
	maxBYOCContinuousProfilingUnixSecond     = 253402300799
)

var (
	ErrInvalidBYOCContinuousProfilingTarget   = errors.New("invalid BYOC Continuous Profiling target")
	ErrInvalidBYOCContinuousProfilingResponse = errors.New("invalid BYOC Continuous Profiling response")
	ErrBYOCContinuousProfilingUnavailable     = errors.New("BYOC Continuous Profiling service unavailable")
	ErrBYOCContinuousProfilingUnsupported     = errors.New("Continuous Profiling is unsupported for this cluster")

	byocContinuousProfilingClusterIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,255}$`)
	byocContinuousProfilingQueryKeys        = map[BYOCContinuousProfilingOperation]map[string]struct{}{
		BYOCContinuousProfilingOperationGroupProfiles: {
			"begin_time": {},
			"end_time":   {},
		},
		BYOCContinuousProfilingOperationGroupProfileDetail: {
			"ts": {},
		},
	}
)

type BYOCContinuousProfilingOperation string

const (
	BYOCContinuousProfilingOperationGroupProfiles      BYOCContinuousProfilingOperation = "group_profiles"
	BYOCContinuousProfilingOperationGroupProfileDetail BYOCContinuousProfilingOperation = "group_profile_detail"
)

type byocContinuousProfilingTargetResolver interface {
	GetBYOCDiagnosisTarget(context.Context, uint64, string) (*clinic_internal_api.BYOCDiagnosisTarget, error)
}

// BYOCContinuousProfilingUpstreamError intentionally excludes the response
// body. O11y errors can contain customer or internal routing details.
type BYOCContinuousProfilingUpstreamError struct {
	StatusCode int
}

func (e *BYOCContinuousProfilingUpstreamError) Error() string {
	return fmt.Sprintf("BYOC Continuous Profiling upstream returned HTTP status %d", e.StatusCode)
}

type BYOCContinuousProfilingResult struct {
	Body         []byte
	Provider     string
	Region       string
	DeployTypeV2 string
}

// BYOCContinuousProfilingService reads the JSON inventory directly from the
// customer-side O11y diagnosis service. This service intentionally has no
// object-storage, warehouse, SQL, or table-manager dependency.
type BYOCContinuousProfilingService struct {
	targetResolver byocContinuousProfilingTargetResolver
	httpClient     *http.Client
}

func NewBYOCContinuousProfilingService(
	targetResolver byocContinuousProfilingTargetResolver,
) *BYOCContinuousProfilingService {
	return newBYOCContinuousProfilingService(targetResolver, &http.Client{
		Transport: clinictracing.NewHTTPTransport(nil),
	})
}

func newBYOCContinuousProfilingService(
	targetResolver byocContinuousProfilingTargetResolver,
	httpClient *http.Client,
) *BYOCContinuousProfilingService {
	if httpClient == nil {
		httpClient = &http.Client{Transport: clinictracing.NewHTTPTransport(nil)}
	}
	clientCopy := *httpClient
	clientCopy.Timeout = byocContinuousProfilingTimeout
	clientCopy.CheckRedirect = func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	}
	return &BYOCContinuousProfilingService{
		targetResolver: targetResolver,
		httpClient:     &clientCopy,
	}
}

func (s *BYOCContinuousProfilingService) Query(
	ctx context.Context,
	orgID uint64,
	clusterID string,
	operation BYOCContinuousProfilingOperation,
	query url.Values,
) (*BYOCContinuousProfilingResult, error) {
	if s == nil || s.targetResolver == nil || s.httpClient == nil {
		return nil, ErrBYOCContinuousProfilingUnavailable
	}
	if err := validateBYOCContinuousProfilingQuery(operation, query); err != nil {
		return nil, err
	}

	ctx, cancel := context.WithTimeout(ctx, byocContinuousProfilingTimeout)
	defer cancel()
	target, err := s.targetResolver.GetBYOCDiagnosisTarget(ctx, orgID, clusterID)
	if err != nil {
		return nil, &byocContinuousProfilingCauseError{
			message: "resolve BYOC Continuous Profiling target failed",
			cause:   err,
		}
	}
	requestURL, err := buildBYOCContinuousProfilingURL(target, orgID, clusterID, operation, query)
	if err != nil {
		return nil, err
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL, nil)
	if err != nil {
		return nil, &byocContinuousProfilingCauseError{
			message: "create BYOC Continuous Profiling request failed",
			cause:   err,
		}
	}
	setBYOCContinuousProfilingHeaders(req, target)

	resp, err := s.httpClient.Do(req)
	if err != nil {
		return nil, &byocContinuousProfilingCauseError{
			message: "call BYOC Continuous Profiling upstream failed",
			cause:   err,
		}
	}
	defer resp.Body.Close()
	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBYOCContinuousProfilingResponse))
		return nil, &BYOCContinuousProfilingUpstreamError{StatusCode: resp.StatusCode}
	}

	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBYOCContinuousProfilingResponse+1))
	if err != nil {
		return nil, &byocContinuousProfilingCauseError{
			message: "read BYOC Continuous Profiling response failed",
			cause:   err,
		}
	}
	if len(body) > maxBYOCContinuousProfilingResponse {
		return nil, fmt.Errorf("%w: response exceeds %d bytes",
			ErrInvalidBYOCContinuousProfilingResponse, maxBYOCContinuousProfilingResponse)
	}
	body, err = validateBYOCContinuousProfilingResponse(operation, body)
	if err != nil {
		return nil, err
	}

	return &BYOCContinuousProfilingResult{
		Body:         body,
		Provider:     target.Provider,
		Region:       target.Region,
		DeployTypeV2: target.DeployTypeV2,
	}, nil
}

func validateBYOCContinuousProfilingQuery(operation BYOCContinuousProfilingOperation, query url.Values) error {
	allowed, ok := byocContinuousProfilingQueryKeys[operation]
	if !ok {
		return errors.New("unsupported BYOC Continuous Profiling operation")
	}
	for key, values := range query {
		if _, ok := allowed[key]; !ok {
			return errors.New("unsupported BYOC Continuous Profiling query parameter")
		}
		if len(values) != 1 {
			return errors.New("BYOC Continuous Profiling query parameters must be scalar")
		}
	}

	switch operation {
	case BYOCContinuousProfilingOperationGroupProfiles:
		begin, err := parseBYOCContinuousProfilingTimestamp(query, "begin_time")
		if err != nil {
			return err
		}
		end, err := parseBYOCContinuousProfilingTimestamp(query, "end_time")
		if err != nil {
			return err
		}
		if end < begin {
			return errors.New("BYOC Continuous Profiling end_time must not precede begin_time")
		}
		begin -= begin % 60
		end -= end % 60
		if end-begin > maxBYOCContinuousProfilingRangeSeconds {
			return errors.New("BYOC Continuous Profiling time range exceeds 3 hours")
		}
	case BYOCContinuousProfilingOperationGroupProfileDetail:
		_, err := parseBYOCContinuousProfilingTimestamp(query, "ts")
		return err
	}
	return nil
}

func parseBYOCContinuousProfilingTimestamp(query url.Values, name string) (int64, error) {
	values, ok := query[name]
	if !ok || len(values) != 1 || values[0] == "" {
		return 0, fmt.Errorf("BYOC Continuous Profiling %s is required", name)
	}
	for _, char := range values[0] {
		if char < '0' || char > '9' {
			return 0, fmt.Errorf("BYOC Continuous Profiling %s must be a valid Unix timestamp", name)
		}
	}
	value, err := strconv.ParseInt(values[0], 10, 64)
	if err != nil || value <= 0 || value > maxBYOCContinuousProfilingUnixSecond {
		return 0, fmt.Errorf("BYOC Continuous Profiling %s must be a valid Unix timestamp", name)
	}
	return value, nil
}

func buildBYOCContinuousProfilingURL(
	target *clinic_internal_api.BYOCDiagnosisTarget,
	orgID uint64,
	clusterID string,
	operation BYOCContinuousProfilingOperation,
	query url.Values,
) (string, error) {
	if err := validateBYOCContinuousProfilingTarget(target, orgID, clusterID); err != nil {
		return "", err
	}
	baseURL, err := url.Parse(target.BaseURL)
	if err != nil {
		return "", fmt.Errorf("%w: baseUrl is invalid", ErrInvalidBYOCContinuousProfilingTarget)
	}

	var suffix string
	switch operation {
	case BYOCContinuousProfilingOperationGroupProfiles:
		suffix = byocContinuousProfilingGroupProfilesPath
	case BYOCContinuousProfilingOperationGroupProfileDetail:
		suffix = byocContinuousProfilingGroupDetailPath
	default:
		return "", errors.New("unsupported BYOC Continuous Profiling operation")
	}
	baseURL.Path = "/internal/" + target.O11yClusterID + suffix
	baseURL.RawPath = ""
	baseURL.RawQuery = query.Encode()
	return baseURL.String(), nil
}

func validateBYOCContinuousProfilingTarget(
	target *clinic_internal_api.BYOCDiagnosisTarget,
	orgID uint64,
	clusterID string,
) error {
	if target == nil {
		return fmt.Errorf("%w: target is empty", ErrInvalidBYOCContinuousProfilingTarget)
	}
	if target.BaseURL == "" || strings.TrimSpace(target.BaseURL) != target.BaseURL {
		return fmt.Errorf("%w: baseUrl is invalid", ErrInvalidBYOCContinuousProfilingTarget)
	}
	baseURL, err := url.Parse(target.BaseURL)
	if err != nil || baseURL.Opaque != "" || (baseURL.Scheme != "http" && baseURL.Scheme != "https") ||
		baseURL.Host == "" || baseURL.Hostname() == "" {
		return fmt.Errorf("%w: baseUrl must be an absolute HTTP(S) URL", ErrInvalidBYOCContinuousProfilingTarget)
	}
	if baseURL.User != nil || baseURL.RawQuery != "" || baseURL.ForceQuery || baseURL.Fragment != "" ||
		baseURL.RawFragment != "" || baseURL.RawPath != "" || (baseURL.Path != "" && baseURL.Path != "/") {
		return fmt.Errorf("%w: baseUrl must contain only scheme and host", ErrInvalidBYOCContinuousProfilingTarget)
	}
	hostname := strings.ToLower(baseURL.Hostname())
	normalizedHostname := strings.TrimSuffix(hostname, ".")
	if normalizedHostname == "localhost" || strings.HasSuffix(normalizedHostname, ".localhost") {
		return fmt.Errorf("%w: localhost is not allowed", ErrInvalidBYOCContinuousProfilingTarget)
	}
	if ip := net.ParseIP(hostname); ip != nil && ip.IsLoopback() {
		return fmt.Errorf("%w: loopback addresses are not allowed", ErrInvalidBYOCContinuousProfilingTarget)
	}
	if !byocContinuousProfilingClusterIDPattern.MatchString(target.O11yClusterID) {
		return fmt.Errorf("%w: o11yClusterId is invalid", ErrInvalidBYOCContinuousProfilingTarget)
	}

	values := []string{
		target.Headers.Env,
		target.Headers.BizType,
		target.Headers.OrgID,
		target.Headers.RegionID,
		target.Headers.ApplicationID,
		target.Provider,
		target.Region,
		target.DeployType,
		target.DeployTypeV2,
	}
	for _, value := range values {
		if value == "" || strings.TrimSpace(value) != value || len(value) > 4096 ||
			containsBYOCContinuousProfilingControlCharacter(value) {
			return fmt.Errorf("%w: routing value is invalid", ErrInvalidBYOCContinuousProfilingTarget)
		}
	}
	if target.Headers.BizType != "byoc" || target.Headers.OrgID != strconv.FormatUint(orgID, 10) ||
		target.Headers.ApplicationID != clusterID || target.Headers.RegionID != target.Provider+"-"+target.Region {
		return fmt.Errorf("%w: target does not match the requested cluster", ErrInvalidBYOCContinuousProfilingTarget)
	}

	if target.DeployType == "byoc" && target.DeployTypeV2 == "byoc" {
		return ErrBYOCContinuousProfilingUnsupported
	}
	if target.DeployType != "byoc-host" || target.DeployTypeV2 != "byoc resource pool" {
		return fmt.Errorf("%w: target is not a physical BYOC cluster", ErrInvalidBYOCContinuousProfilingTarget)
	}
	return nil
}

func containsBYOCContinuousProfilingControlCharacter(value string) bool {
	for _, r := range value {
		if unicode.IsControl(r) {
			return true
		}
	}
	return false
}

func setBYOCContinuousProfilingHeaders(
	req *http.Request,
	target *clinic_internal_api.BYOCDiagnosisTarget,
) {
	req.Header.Set("Accept", "application/json")
	req.Header.Set("X-Env", target.Headers.Env)
	req.Header.Set("X-Biz-Type", target.Headers.BizType)
	req.Header.Set("X-Org-Id", target.Headers.OrgID)
	req.Header.Set("X-Region-Id", target.Headers.RegionID)
	req.Header.Set("X-Application-Id", target.Headers.ApplicationID)
	req.Header.Set("X-Provider", target.Provider)
	req.Header.Set("X-Region", target.Region)
	req.Header.Set("X-Cluster-Id", target.Headers.ApplicationID)
	req.Header.Set("X-Deploy-Type", target.DeployType)
}

func validateBYOCContinuousProfilingResponse(
	operation BYOCContinuousProfilingOperation,
	body []byte,
) ([]byte, error) {
	trimmed := bytes.TrimSpace(body)
	if !jsonValueIsValid(trimmed) {
		return nil, fmt.Errorf("%w: response is not valid JSON", ErrInvalidBYOCContinuousProfilingResponse)
	}
	switch operation {
	case BYOCContinuousProfilingOperationGroupProfiles:
		if bytes.Equal(trimmed, []byte("null")) {
			return []byte("[]"), nil
		}
		if len(trimmed) == 0 || trimmed[0] != '[' {
			return nil, fmt.Errorf("%w: group profiles response must be an array",
				ErrInvalidBYOCContinuousProfilingResponse)
		}
	case BYOCContinuousProfilingOperationGroupProfileDetail:
		if !bytes.Equal(trimmed, []byte("null")) && (len(trimmed) == 0 || trimmed[0] != '{') {
			return nil, fmt.Errorf("%w: group profile detail response must be an object or null",
				ErrInvalidBYOCContinuousProfilingResponse)
		}
	default:
		return nil, errors.New("unsupported BYOC Continuous Profiling operation")
	}
	return body, nil
}

func jsonValueIsValid(value []byte) bool {
	return len(value) != 0 && json.Valid(value)
}

type byocContinuousProfilingCauseError struct {
	message string
	cause   error
}

func (e *byocContinuousProfilingCauseError) Error() string {
	return e.message
}

func (e *byocContinuousProfilingCauseError) Unwrap() error {
	return e.cause
}