package service

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

	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 (
	byocSlowQueryAvailableFieldsPath = "/diagnosis/api/v1/slow_query/available_fields"
	byocSlowQueryListPath            = "/diagnosis/api/v1/slow_query/list"
	byocSlowQueryDetailPath          = "/diagnosis/api/v1/slow_query/detail"
	maxBYOCSlowQueryResponse         = 16 << 20
)

var (
	ErrInvalidBYOCSlowQueryTarget   = errors.New("invalid BYOC Slow Query target")
	ErrInvalidBYOCSlowQueryResponse = errors.New("invalid BYOC Slow Query response")
	ErrBYOCSlowQueryUnavailable     = errors.New("BYOC Slow Query service unavailable")

	byocSlowQueryClusterIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,255}$`)

	// O11y always selects these columns for a list row because they form the
	// composite key accepted by the detail endpoint. Keep them in the
	// canonical response even when the caller did not explicitly request them.
	byocSlowQueryListKeyFields = []string{"digest", "timestamp", "connection_id"}
)

type BYOCSlowQueryOperation string

const (
	BYOCSlowQueryOperationAvailableFields BYOCSlowQueryOperation = "available_fields"
	BYOCSlowQueryOperationList            BYOCSlowQueryOperation = "list"
	BYOCSlowQueryOperationDetail          BYOCSlowQueryOperation = "detail"
)

type byocSlowQueryTargetResolver interface {
	GetSlowQueryTarget(context.Context, uint64, string) (*clinic_internal_api.SlowQueryTarget, error)
}

// BYOCSlowQueryUpstreamError intentionally contains only the upstream status.
// Customer query text and upstream response bodies must not escape the proxy.
type BYOCSlowQueryUpstreamError struct {
	StatusCode int
}

func (e *BYOCSlowQueryUpstreamError) Error() string {
	return fmt.Sprintf("BYOC Slow Query upstream returned HTTP status %d", e.StatusCode)
}

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

// BYOCSlowQueryService calls only fixed O11y diagnosis endpoints. It has no
// dependency on the Data Proxy SQL engine, object storage, or table managers.
type BYOCSlowQueryService struct {
	targetResolver byocSlowQueryTargetResolver
	httpClient     *http.Client
}

func NewBYOCSlowQueryService(targetResolver byocSlowQueryTargetResolver) *BYOCSlowQueryService {
	return newBYOCSlowQueryService(targetResolver, &http.Client{
		Timeout:   2 * time.Minute,
		Transport: clinictracing.NewHTTPTransport(nil),
	})
}

func newBYOCSlowQueryService(targetResolver byocSlowQueryTargetResolver, httpClient *http.Client) *BYOCSlowQueryService {
	if httpClient == nil {
		httpClient = &http.Client{
			Timeout:   2 * time.Minute,
			Transport: clinictracing.NewHTTPTransport(nil),
		}
	}
	clientCopy := *httpClient
	clientCopy.CheckRedirect = func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	}
	return &BYOCSlowQueryService{targetResolver: targetResolver, httpClient: &clientCopy}
}

func (s *BYOCSlowQueryService) Query(
	ctx context.Context,
	orgID uint64,
	clusterID string,
	operation BYOCSlowQueryOperation,
	query url.Values,
) (*BYOCSlowQueryResult, error) {
	if s == nil || s.targetResolver == nil {
		return nil, ErrBYOCSlowQueryUnavailable
	}
	target, err := s.targetResolver.GetSlowQueryTarget(ctx, orgID, clusterID)
	if err != nil {
		return nil, fmt.Errorf("resolve BYOC Slow Query target: %w", err)
	}
	requestURL, err := buildBYOCSlowQueryURL(target, operation, query)
	if err != nil {
		return nil, err
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL, nil)
	if err != nil {
		return nil, fmt.Errorf("create BYOC Slow Query request: %w", err)
	}
	req.Header.Set("Accept", "application/json")
	setBYOCSlowQueryHeader(req.Header, "X-Env", target.Headers.Env)
	setBYOCSlowQueryHeader(req.Header, "X-Biz-Type", target.Headers.BizType)
	setBYOCSlowQueryHeader(req.Header, "X-Org-Id", target.Headers.OrgID)
	setBYOCSlowQueryHeader(req.Header, "X-Region-Id", target.Headers.RegionID)
	setBYOCSlowQueryHeader(req.Header, "X-Application-Id", target.Headers.ApplicationID)

	resp, err := s.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("call BYOC Slow Query upstream: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBYOCSlowQueryResponse))
		return nil, &BYOCSlowQueryUpstreamError{StatusCode: resp.StatusCode}
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBYOCSlowQueryResponse+1))
	if err != nil {
		return nil, fmt.Errorf("read BYOC Slow Query response: %w", err)
	}
	if len(body) > maxBYOCSlowQueryResponse {
		return nil, fmt.Errorf("%w: response exceeds %d bytes", ErrInvalidBYOCSlowQueryResponse, maxBYOCSlowQueryResponse)
	}
	if !isValidBYOCSlowQueryJSON(body, operation) {
		return nil, fmt.Errorf("%w: response has an invalid JSON shape", ErrInvalidBYOCSlowQueryResponse)
	}
	if operation == BYOCSlowQueryOperationList && query.Get("fields") != "" {
		body, err = projectBYOCSlowQueryList(body, query.Get("fields"))
		if err != nil {
			return nil, err
		}
	}

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

// projectBYOCSlowQueryList enforces the canonical fields contract at the
// proxy boundary. The O11y/TiDB Dashboard implementation limits its SQL
// projection, but marshals the result through a full struct without omitempty;
// consequently unrequested fields (including query and plan) otherwise appear
// as zero-value JSON keys. RawMessage keeps number lexemes unchanged while the
// object is projected.
func projectBYOCSlowQueryList(body []byte, fields string) ([]byte, error) {
	var rows []map[string]json.RawMessage
	decoder := json.NewDecoder(bytes.NewReader(body))
	if err := decoder.Decode(&rows); err != nil {
		return nil, fmt.Errorf("%w: decode list response", ErrInvalidBYOCSlowQueryResponse)
	}

	allowed := make(map[string]struct{})
	for _, field := range strings.Split(fields, ",") {
		field = strings.ToLower(strings.TrimSpace(field))
		if field != "" {
			allowed[field] = struct{}{}
		}
	}
	for _, field := range byocSlowQueryListKeyFields {
		allowed[field] = struct{}{}
	}

	projected := make([]map[string]json.RawMessage, 0, len(rows))
	for _, row := range rows {
		if row == nil {
			return nil, fmt.Errorf("%w: list row must be an object", ErrInvalidBYOCSlowQueryResponse)
		}
		item := make(map[string]json.RawMessage, len(allowed)+1)
		for key, value := range row {
			if _, ok := allowed[strings.ToLower(key)]; ok {
				item[key] = value
			}
		}

		projected = append(projected, item)
	}

	result, err := json.Marshal(projected)
	if err != nil {
		return nil, fmt.Errorf("%w: encode projected list response", ErrInvalidBYOCSlowQueryResponse)
	}
	if len(result) > maxBYOCSlowQueryResponse {
		return nil, fmt.Errorf("%w: projected response exceeds %d bytes", ErrInvalidBYOCSlowQueryResponse, maxBYOCSlowQueryResponse)
	}
	return result, nil
}

func buildBYOCSlowQueryURL(
	target *clinic_internal_api.SlowQueryTarget,
	operation BYOCSlowQueryOperation,
	query url.Values,
) (string, error) {
	if target == nil {
		return "", fmt.Errorf("%w: target is empty", ErrInvalidBYOCSlowQueryTarget)
	}
	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", ErrInvalidBYOCSlowQueryTarget)
	}
	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", ErrInvalidBYOCSlowQueryTarget)
	}
	hostname := strings.ToLower(baseURL.Hostname())
	normalizedHostname := strings.TrimSuffix(hostname, ".")
	if normalizedHostname == "localhost" || strings.HasSuffix(normalizedHostname, ".localhost") {
		return "", fmt.Errorf("%w: localhost is not allowed", ErrInvalidBYOCSlowQueryTarget)
	}
	if ip := net.ParseIP(hostname); ip != nil && ip.IsLoopback() {
		return "", fmt.Errorf("%w: loopback addresses are not allowed", ErrInvalidBYOCSlowQueryTarget)
	}
	if !byocSlowQueryClusterIDPattern.MatchString(target.O11yClusterID) {
		return "", fmt.Errorf("%w: o11yClusterId is invalid", ErrInvalidBYOCSlowQueryTarget)
	}

	var suffix string
	switch operation {
	case BYOCSlowQueryOperationAvailableFields:
		suffix = byocSlowQueryAvailableFieldsPath
	case BYOCSlowQueryOperationList:
		suffix = byocSlowQueryListPath
	case BYOCSlowQueryOperationDetail:
		suffix = byocSlowQueryDetailPath
	default:
		return "", fmt.Errorf("unsupported BYOC Slow Query operation %q", operation)
	}
	baseURL.Path = "/internal/" + target.O11yClusterID + suffix
	baseURL.RawPath = ""
	baseURL.RawQuery = query.Encode()
	return baseURL.String(), nil
}

func setBYOCSlowQueryHeader(header http.Header, name, value string) {
	if value != "" {
		header.Set(name, value)
	}
}

func isValidBYOCSlowQueryJSON(body []byte, operation BYOCSlowQueryOperation) bool {
	trimmed := bytes.TrimSpace(body)
	if len(trimmed) == 0 || !json.Valid(trimmed) {
		return false
	}
	switch operation {
	case BYOCSlowQueryOperationAvailableFields, BYOCSlowQueryOperationList:
		return trimmed[0] == '['
	case BYOCSlowQueryOperationDetail:
		return trimmed[0] == '{'
	default:
		return false
	}
}