package service

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"regexp"
	"strconv"
	"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 (
	byocStatementsAvailableFieldsPath = "/diagnosis/api/v1/statements/available_fields"
	byocStatementsTypesPath           = "/diagnosis/api/v1/statements/stmt_types"
	byocStatementsListPath            = "/diagnosis/api/v1/statements/list"
	byocStatementsPlansPath           = "/diagnosis/api/v1/statements/plans"
	byocStatementsPlanDetailPath      = "/diagnosis/api/v1/statements/plan/detail"
	maxBYOCStatementsResponse         = 16 << 20
	byocStatementsTimeout             = 2 * time.Minute
)

var (
	ErrInvalidBYOCStatementsTarget   = errors.New("invalid BYOC Statements target")
	ErrInvalidBYOCStatementsResponse = errors.New("invalid BYOC Statements response")
	ErrBYOCStatementsUnavailable     = errors.New("BYOC Statements service unavailable")

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

type BYOCStatementsOperation string

const (
	BYOCStatementsOperationAvailableFields BYOCStatementsOperation = "available_fields"
	BYOCStatementsOperationTypes           BYOCStatementsOperation = "statement_types"
	BYOCStatementsOperationList            BYOCStatementsOperation = "list"
	BYOCStatementsOperationPlans           BYOCStatementsOperation = "plans"
	BYOCStatementsOperationPlanDetail      BYOCStatementsOperation = "plan_detail"
)

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

// BYOCStatementsUpstreamError intentionally carries only an HTTP status. The
// upstream body may contain customer SQL and must never escape this proxy.
type BYOCStatementsUpstreamError struct {
	StatusCode int
}

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

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

// BYOCStatementsService calls only fixed customer-side O11y diagnosis
// endpoints. It deliberately has no SQL engine, object-storage, warehouse,
// table-manager, cache, or persistence dependency.
type BYOCStatementsService struct {
	targetResolver byocStatementsTargetResolver
	httpClient     *http.Client
}

func NewBYOCStatementsService(targetResolver byocStatementsTargetResolver) *BYOCStatementsService {
	return newBYOCStatementsService(targetResolver, &http.Client{
		Timeout:   byocStatementsTimeout,
		Transport: clinictracing.NewHTTPTransport(nil),
	})
}

func newBYOCStatementsService(targetResolver byocStatementsTargetResolver, httpClient *http.Client) *BYOCStatementsService {
	if httpClient == nil {
		httpClient = &http.Client{
			Timeout:   byocStatementsTimeout,
			Transport: clinictracing.NewHTTPTransport(nil),
		}
	}
	clientCopy := *httpClient
	clientCopy.CheckRedirect = func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	}
	return &BYOCStatementsService{targetResolver: targetResolver, httpClient: &clientCopy}
}

func (s *BYOCStatementsService) Query(
	ctx context.Context,
	orgID uint64,
	clusterID string,
	operation BYOCStatementsOperation,
	query url.Values,
) (*BYOCStatementsResult, error) {
	// Public callers must pass the canonicalized values produced by the strict
	// BYOCStatementsHandler parsers. This service still rechecks the operation's
	// key allowlist so a future internal caller cannot select a storage backend
	// or forward a new O11y parameter implicitly.
	if s == nil || s.targetResolver == nil {
		return nil, ErrBYOCStatementsUnavailable
	}
	if err := validateBYOCStatementsQuery(operation, query); err != nil {
		return nil, err
	}

	ctx, cancel := context.WithTimeout(ctx, byocStatementsTimeout)
	defer cancel()

	target, err := s.targetResolver.GetBYOCDiagnosisTarget(ctx, orgID, clusterID)
	if err != nil {
		return nil, fmt.Errorf("resolve BYOC Statements target: %w", err)
	}
	if err := validateBYOCStatementsTargetScope(target, orgID, clusterID); err != nil {
		return nil, err
	}
	requestURL, err := buildBYOCStatementsURL(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 Statements request: %w", err)
	}
	req.Header.Set("Accept", "application/json")
	setBYOCStatementsHeader(req.Header, "X-Env", target.Headers.Env)
	setBYOCStatementsHeader(req.Header, "X-Biz-Type", target.Headers.BizType)
	setBYOCStatementsHeader(req.Header, "X-Org-Id", target.Headers.OrgID)
	setBYOCStatementsHeader(req.Header, "X-Region-Id", target.Headers.RegionID)
	setBYOCStatementsHeader(req.Header, "X-Application-Id", target.Headers.ApplicationID)
	setBYOCStatementsHeader(req.Header, "X-Provider", target.Provider)
	setBYOCStatementsHeader(req.Header, "X-Region", target.Region)
	setBYOCStatementsHeader(req.Header, "X-Cluster-Id", target.Headers.ApplicationID)
	setBYOCStatementsHeader(req.Header, "X-Deploy-Type", target.DeployType)

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

	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBYOCStatementsResponse))
		return nil, &BYOCStatementsUpstreamError{StatusCode: resp.StatusCode}
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBYOCStatementsResponse+1))
	if err != nil {
		return nil, fmt.Errorf("read BYOC Statements response: %w", err)
	}
	if len(body) > maxBYOCStatementsResponse {
		return nil, fmt.Errorf("%w: response exceeds %d bytes", ErrInvalidBYOCStatementsResponse, maxBYOCStatementsResponse)
	}
	body = normalizeBYOCStatementsEmptyCollection(body, operation)
	if !isValidBYOCStatementsJSON(body, operation) {
		return nil, fmt.Errorf("%w: response has an invalid JSON shape", ErrInvalidBYOCStatementsResponse)
	}
	if operation == BYOCStatementsOperationPlanDetail {
		body = addBYOCStatementsBinaryPlanText(body)
		if len(body) > maxBYOCStatementsResponse {
			return nil, fmt.Errorf("%w: enriched response exceeds %d bytes", ErrInvalidBYOCStatementsResponse, maxBYOCStatementsResponse)
		}
	}

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

func validateBYOCStatementsTargetScope(target *clinic_internal_api.BYOCDiagnosisTarget, orgID uint64, clusterID string) error {
	if target == nil {
		return fmt.Errorf("%w: target is empty", ErrInvalidBYOCStatementsTarget)
	}
	required := []string{
		target.BaseURL,
		target.O11yClusterID,
		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 required {
		if value == "" || strings.TrimSpace(value) != value {
			return fmt.Errorf("%w: target contains an empty or malformed field", ErrInvalidBYOCStatementsTarget)
		}
	}
	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 scope does not match the request", ErrInvalidBYOCStatementsTarget)
	}
	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", ErrInvalidBYOCStatementsTarget)
	}
	return nil
}

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

	var suffix string
	switch operation {
	case BYOCStatementsOperationAvailableFields:
		suffix = byocStatementsAvailableFieldsPath
	case BYOCStatementsOperationTypes:
		suffix = byocStatementsTypesPath
	case BYOCStatementsOperationList:
		suffix = byocStatementsListPath
	case BYOCStatementsOperationPlans:
		suffix = byocStatementsPlansPath
	case BYOCStatementsOperationPlanDetail:
		suffix = byocStatementsPlanDetailPath
	default:
		return "", fmt.Errorf("unsupported BYOC Statements operation %q", operation)
	}
	baseURL.Path = "/internal/" + target.O11yClusterID + suffix
	baseURL.RawPath = ""
	baseURL.RawQuery = query.Encode()
	return baseURL.String(), nil
}

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

func validateBYOCStatementsQuery(operation BYOCStatementsOperation, query url.Values) error {
	allowed := map[string]struct{}{}
	switch operation {
	case BYOCStatementsOperationAvailableFields, BYOCStatementsOperationTypes:
	case BYOCStatementsOperationList:
		allowed = map[string]struct{}{
			"begin_time": {}, "end_time": {}, "schemas": {}, "stmt_types": {}, "text": {}, "fields": {},
		}
	case BYOCStatementsOperationPlans:
		allowed = map[string]struct{}{
			"begin_time": {}, "end_time": {}, "digest": {}, "schema_name": {},
		}
	case BYOCStatementsOperationPlanDetail:
		allowed = map[string]struct{}{
			"begin_time": {}, "end_time": {}, "digest": {}, "plans": {}, "schema_name": {},
		}
	default:
		return fmt.Errorf("unsupported BYOC Statements operation %q", operation)
	}
	for key := range query {
		if _, ok := allowed[key]; !ok {
			return fmt.Errorf("unsupported BYOC Statements parameter %q", key)
		}
	}
	return nil
}

func normalizeBYOCStatementsEmptyCollection(body []byte, operation BYOCStatementsOperation) []byte {
	switch operation {
	case BYOCStatementsOperationList, BYOCStatementsOperationPlans:
		if bytes.Equal(bytes.TrimSpace(body), []byte("null")) {
			return []byte("[]")
		}
	}
	return body
}

func isValidBYOCStatementsJSON(body []byte, operation BYOCStatementsOperation) bool {
	trimmed := bytes.TrimSpace(body)
	if len(trimmed) == 0 || !json.Valid(trimmed) {
		return false
	}

	switch operation {
	case BYOCStatementsOperationAvailableFields, BYOCStatementsOperationTypes:
		if trimmed[0] != '[' {
			return false
		}
		var items []json.RawMessage
		if json.Unmarshal(trimmed, &items) != nil {
			return false
		}
		for _, item := range items {
			item = bytes.TrimSpace(item)
			if len(item) == 0 || item[0] != '"' {
				return false
			}
			var value string
			if json.Unmarshal(item, &value) != nil {
				return false
			}
		}
		return true
	case BYOCStatementsOperationList, BYOCStatementsOperationPlans:
		if trimmed[0] != '[' {
			return false
		}
		var items []json.RawMessage
		if json.Unmarshal(trimmed, &items) != nil {
			return false
		}
		for _, item := range items {
			item = bytes.TrimSpace(item)
			if len(item) == 0 || item[0] != '{' {
				return false
			}
		}
		return true
	case BYOCStatementsOperationPlanDetail:
		if trimmed[0] != '{' {
			return false
		}
		var object map[string]json.RawMessage
		return json.Unmarshal(trimmed, &object) == nil && object != nil
	default:
		return false
	}
}