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 (
	byocTopSQLInstancesPath = "/diagnosis/api/v1/topsql/instances"
	byocTopSQLSummaryPath   = "/diagnosis/api/v1/topsql/summary"
	maxBYOCTopSQLResponse   = 16 << 20
	byocTopSQLTimeout       = 2 * time.Minute
)

var (
	ErrInvalidBYOCTopSQLTarget   = errors.New("invalid BYOC Top SQL target")
	ErrInvalidBYOCTopSQLResponse = errors.New("invalid BYOC Top SQL response")
	ErrBYOCTopSQLUnavailable     = errors.New("BYOC Top SQL service unavailable")

	byocTopSQLClusterIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,255}$`)
	byocTopSQLQueryKeys        = map[BYOCTopSQLOperation]map[string]struct{}{
		BYOCTopSQLOperationInstances: {
			"start": {},
			"end":   {},
		},
		BYOCTopSQLOperationSummary: {
			"start":         {},
			"end":           {},
			"window":        {},
			"top":           {},
			"instance":      {},
			"instance_type": {},
		},
	}
)

type BYOCTopSQLOperation string

const (
	BYOCTopSQLOperationInstances BYOCTopSQLOperation = "instances"
	BYOCTopSQLOperationSummary   BYOCTopSQLOperation = "summary"
)

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

// BYOCTopSQLUpstreamError intentionally contains only the status code. O11y
// response bodies can contain implementation or customer details and must not
// escape the proxy.
type BYOCTopSQLUpstreamError struct {
	StatusCode int
}

func (e *BYOCTopSQLUpstreamError) Error() string {
	return fmt.Sprintf("BYOC Top SQL upstream returned HTTP status %d", e.StatusCode)
}

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

// BYOCTopSQLService reads Top SQL directly from the fixed O11y diagnosis
// endpoints. It has no dependency on Data Proxy SQL, object storage, warehouse
// tables, or table managers.
type BYOCTopSQLService struct {
	targetResolver byocTopSQLTargetResolver
	httpClient     *http.Client
}

func NewBYOCTopSQLService(targetResolver byocTopSQLTargetResolver) *BYOCTopSQLService {
	return newBYOCTopSQLService(targetResolver, &http.Client{
		Transport: clinictracing.NewHTTPTransport(nil),
	})
}

func newBYOCTopSQLService(targetResolver byocTopSQLTargetResolver, httpClient *http.Client) *BYOCTopSQLService {
	if httpClient == nil {
		httpClient = &http.Client{Transport: clinictracing.NewHTTPTransport(nil)}
	}
	clientCopy := *httpClient
	clientCopy.Timeout = byocTopSQLTimeout
	clientCopy.CheckRedirect = func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	}
	return &BYOCTopSQLService{targetResolver: targetResolver, httpClient: &clientCopy}
}

func (s *BYOCTopSQLService) Query(
	ctx context.Context,
	orgID uint64,
	clusterID string,
	operation BYOCTopSQLOperation,
	query url.Values,
) (*BYOCTopSQLResult, error) {
	if s == nil || s.targetResolver == nil || s.httpClient == nil {
		return nil, ErrBYOCTopSQLUnavailable
	}
	ctx, cancel := context.WithTimeout(ctx, byocTopSQLTimeout)
	defer cancel()
	if err := validateBYOCTopSQLQuery(operation, query); err != nil {
		return nil, err
	}

	target, err := s.targetResolver.GetBYOCDiagnosisTarget(ctx, orgID, clusterID)
	if err != nil {
		return nil, &byocTopSQLCauseError{message: "resolve BYOC Top SQL target failed", cause: err}
	}
	requestURL, err := buildBYOCTopSQLURL(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, &byocTopSQLCauseError{message: "create BYOC Top SQL request failed", cause: err}
	}
	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)

	resp, err := s.httpClient.Do(req)
	if err != nil {
		return nil, &byocTopSQLCauseError{message: "call BYOC Top SQL 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, maxBYOCTopSQLResponse))
		return nil, &BYOCTopSQLUpstreamError{StatusCode: resp.StatusCode}
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBYOCTopSQLResponse+1))
	if err != nil {
		return nil, &byocTopSQLCauseError{message: "read BYOC Top SQL response failed", cause: err}
	}
	if len(body) > maxBYOCTopSQLResponse {
		return nil, fmt.Errorf("%w: response exceeds %d bytes", ErrInvalidBYOCTopSQLResponse, maxBYOCTopSQLResponse)
	}
	if !isValidBYOCTopSQLEnvelope(body) {
		return nil, fmt.Errorf("%w: response has an invalid JSON envelope", ErrInvalidBYOCTopSQLResponse)
	}

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

func validateBYOCTopSQLQuery(operation BYOCTopSQLOperation, query url.Values) error {
	allowed, ok := byocTopSQLQueryKeys[operation]
	if !ok {
		return errors.New("unsupported BYOC Top SQL operation")
	}
	for key := range query {
		if _, ok := allowed[key]; !ok {
			// In particular, data_source=deltalake must never reach O11y for a
			// BYOC cluster because the customer remains the data owner.
			return errors.New("unsupported BYOC Top SQL query parameter")
		}
	}
	return nil
}

func buildBYOCTopSQLURL(
	target *clinic_internal_api.BYOCDiagnosisTarget,
	orgID uint64,
	clusterID string,
	operation BYOCTopSQLOperation,
	query url.Values,
) (string, error) {
	if err := validateBYOCTopSQLTarget(target, orgID, clusterID); err != nil {
		return "", err
	}

	baseURL, err := url.Parse(target.BaseURL)
	if err != nil {
		return "", fmt.Errorf("%w: baseUrl is invalid", ErrInvalidBYOCTopSQLTarget)
	}
	var suffix string
	switch operation {
	case BYOCTopSQLOperationInstances:
		suffix = byocTopSQLInstancesPath
	case BYOCTopSQLOperationSummary:
		suffix = byocTopSQLSummaryPath
	default:
		return "", errors.New("unsupported BYOC Top SQL operation")
	}
	baseURL.Path = "/internal/" + target.O11yClusterID + suffix
	baseURL.RawPath = ""
	baseURL.RawQuery = query.Encode()
	return baseURL.String(), nil
}

func validateBYOCTopSQLTarget(target *clinic_internal_api.BYOCDiagnosisTarget, orgID uint64, clusterID string) error {
	if target == nil {
		return fmt.Errorf("%w: target is empty", ErrInvalidBYOCTopSQLTarget)
	}
	if target.BaseURL == "" || strings.TrimSpace(target.BaseURL) != target.BaseURL {
		return fmt.Errorf("%w: baseUrl is invalid", ErrInvalidBYOCTopSQLTarget)
	}
	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", ErrInvalidBYOCTopSQLTarget)
	}
	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", ErrInvalidBYOCTopSQLTarget)
	}
	hostname := strings.ToLower(baseURL.Hostname())
	normalizedHostname := strings.TrimSuffix(hostname, ".")
	if normalizedHostname == "localhost" || strings.HasSuffix(normalizedHostname, ".localhost") {
		return fmt.Errorf("%w: localhost is not allowed", ErrInvalidBYOCTopSQLTarget)
	}
	if ip := net.ParseIP(hostname); ip != nil && ip.IsLoopback() {
		return fmt.Errorf("%w: loopback addresses are not allowed", ErrInvalidBYOCTopSQLTarget)
	}
	if !byocTopSQLClusterIDPattern.MatchString(target.O11yClusterID) {
		return fmt.Errorf("%w: o11yClusterId is invalid", ErrInvalidBYOCTopSQLTarget)
	}

	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 || containsControlCharacter(value) {
			return fmt.Errorf("%w: routing value is invalid", ErrInvalidBYOCTopSQLTarget)
		}
	}
	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", ErrInvalidBYOCTopSQLTarget)
	}
	switch {
	case target.DeployType == "byoc" && target.DeployTypeV2 == "byoc":
	case target.DeployType == "byoc-host" && target.DeployTypeV2 == "byoc resource pool":
	default:
		return fmt.Errorf("%w: deploy type is not a consistent BYOC target", ErrInvalidBYOCTopSQLTarget)
	}
	return nil
}

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

func isValidBYOCTopSQLEnvelope(body []byte) bool {
	decoder := json.NewDecoder(bytes.NewReader(body))
	decoder.UseNumber()
	var envelope struct {
		Status json.RawMessage `json:"status"`
		Data   json.RawMessage `json:"data"`
	}
	if err := decoder.Decode(&envelope); err != nil {
		return false
	}
	var trailing any
	if decoder.Decode(&trailing) != io.EOF {
		return false
	}

	if len(envelope.Status) == 0 {
		return false
	}
	var status string
	if err := json.Unmarshal(envelope.Status, &status); err != nil || status == "" {
		return false
	}
	if len(envelope.Data) == 0 {
		return false
	}
	trimmedData := bytes.TrimSpace(envelope.Data)
	if len(trimmedData) == 0 || trimmedData[0] != '[' {
		return false
	}
	// The top-level decoder has already validated the JSON value. Avoid
	// unmarshalling a potentially large data array into one RawMessage per item,
	// which would amplify memory for an otherwise bounded response.
	return json.Valid(trimmedData)
}

// byocTopSQLCauseError preserves errors.Is/errors.As for handler mapping while
// keeping target URLs, query parameters, and lower-level error text out of
// logs and public error strings.
type byocTopSQLCauseError struct {
	message string
	cause   error
}

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

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