package service

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

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	clinictracing "github.com/pingcap-inc/clinic/clinic_data_service/pkg/tracing"
	"go.uber.org/zap"
)

// LokiClusterInfo is the subset of cluster data needed for LogQL label injection by deployTypeV2.
type LokiClusterInfo struct {
	ClusterID         string
	TenantID          string
	Provider          string
	Region            string
	DeployType        string
	ParentID          string
	DeployTypeV2      string
	ProjectID         string
	ClusterMetricsURL string
	LokiURL           string
}

// ErrInvalidBYOCLogQuery identifies caller-controlled LogQL that cannot be
// safely constrained to one BYOC cluster.
var ErrInvalidBYOCLogQuery = errors.New("invalid BYOC Loki query")

// LokiProxyService handles Loki API proxy requests
type LokiProxyService struct {
	httpClient              *http.Client
	env                     string
	serverlessParentEnv     string
	maxLimit                int // Maximum limit for Loki queries
	defaultSystemDatasource string
	systemEndpoints         []config.SystemLokiEndpointConfig
}

const (
	metricsGatewayConnectionSlaTotal = "gateway_connection_sla_total"
	labelToken                       = "token"
)

// NewLokiProxyService creates a new LokiProxyService
func NewLokiProxyService(cfg *config.DataProxyConfig) *LokiProxyService {
	maxLimit := cfg.LokiProxy.MaxLimit
	if maxLimit <= 0 {
		maxLimit = 10000 // Default max limit
	}

	env := cfg.LokiProxy.Env
	if env == "" {
		env = "dev"
	}

	return &LokiProxyService{
		httpClient: &http.Client{
			Timeout:   2 * time.Minute,
			Transport: clinictracing.NewHTTPTransport(nil),
		},
		env:                     env,
		serverlessParentEnv:     serverlessParentMetricsEnv(cfg),
		maxLimit:                maxLimit,
		defaultSystemDatasource: cfg.LokiProxy.DefaultSystemDatasource,
		systemEndpoints:         cfg.LokiProxy.SystemEndpoints,
	}
}

type SystemLogsDatasource struct {
	Datasource string `json:"datasource"`
	UID        string `json:"uid,omitempty"`
	Provider   string `json:"provider,omitempty"`
	Region     string `json:"region,omitempty"`
	IsDefault  bool   `json:"is_default"`
}

type SystemLogsDatasources struct {
	DefaultDatasource string                 `json:"default_datasource,omitempty"`
	Datasources       []SystemLogsDatasource `json:"datasources"`
}

// ListSystemLogDatasources returns configured system log datasources without exposing upstream URLs.
func (s *LokiProxyService) ListSystemLogDatasources() SystemLogsDatasources {
	defaultDatasource := strings.TrimSpace(s.defaultSystemDatasource)
	datasources := make([]SystemLogsDatasource, 0, len(s.systemEndpoints))
	for _, endpoint := range s.systemEndpoints {
		datasource := systemLogsDatasource(endpoint)
		if datasource == "" {
			continue
		}
		datasources = append(datasources, SystemLogsDatasource{
			Datasource: datasource,
			UID:        endpoint.UID,
			Provider:   endpoint.Provider,
			Region:     endpoint.Region,
			IsDefault:  defaultDatasource != "" && strings.EqualFold(datasource, defaultDatasource),
		})
	}
	return SystemLogsDatasources{
		DefaultDatasource: defaultDatasource,
		Datasources:       datasources,
	}
}

// ResolveSystemLogsURL returns the configured control-plane Loki URL for a datasource.
func (s *LokiProxyService) ResolveSystemLogsURL(datasource string) (string, error) {
	datasource = strings.TrimSpace(datasource)
	if len(s.systemEndpoints) == 0 {
		return "", fmt.Errorf("system log datasources are not configured")
	}
	if datasource == "" {
		return "", fmt.Errorf("datasource is required")
	}
	for _, endpoint := range s.systemEndpoints {
		if strings.EqualFold(systemLogsDatasource(endpoint), datasource) && strings.TrimSpace(endpoint.URL) != "" {
			return endpoint.URL, nil
		}
	}
	return "", fmt.Errorf("system log datasource not found: %s", datasource)
}

// ResolveSystemLogsURLByProviderRegion returns the configured control-plane Loki URL for a provider/region pair.
func (s *LokiProxyService) ResolveSystemLogsURLByProviderRegion(provider, region string) (string, error) {
	provider = strings.TrimSpace(provider)
	region = strings.TrimSpace(region)
	if len(s.systemEndpoints) == 0 {
		return "", fmt.Errorf("system log datasources are not configured")
	}
	if provider == "" || region == "" {
		return "", fmt.Errorf("provider and region are required")
	}
	for _, endpoint := range s.systemEndpoints {
		if strings.EqualFold(strings.TrimSpace(endpoint.Provider), provider) &&
			strings.EqualFold(strings.TrimSpace(endpoint.Region), region) &&
			strings.TrimSpace(endpoint.URL) != "" {
			return endpoint.URL, nil
		}
	}
	return "", fmt.Errorf("system log datasource not found for provider=%s region=%s", provider, region)
}

// ResolveServerlessParentLogsURL routes parent/control-plane logs by datasource override or control_plane_info provider/region.
func (s *LokiProxyService) ResolveServerlessParentLogsURL(datasource string, info ServerlessParentControlPlaneInfo) (string, error) {
	if strings.TrimSpace(datasource) != "" {
		return s.ResolveSystemLogsURL(datasource)
	}
	return s.ResolveSystemLogsURLByProviderRegion(info.Vendor, info.Region)
}

func (s *LokiProxyService) serverlessParentLogsClusterLabel(info ServerlessParentControlPlaneInfo) string {
	env := strings.TrimSpace(s.serverlessParentEnv)
	if env == "" {
		env = "dev"
	}
	return fmt.Sprintf("%s-%s", env, strings.TrimSpace(info.ProjectID))
}

func systemLogsDatasource(endpoint config.SystemLokiEndpointConfig) string {
	if strings.TrimSpace(endpoint.Datasource) != "" {
		return endpoint.Datasource
	}
	return endpoint.UID
}

// proxyPrometheusSeriesResult parses Prometheus /api/v1/series response: {"status":"success","data":[{"__name__":"...", "label":"value", ...}, ...]}
type proxyPrometheusSeriesResult struct {
	Data []map[string]string `json:"data"`
}

const fetchKeyspaceNameDefaultRange = 24 * time.Hour

// fetchMetricsLabel queries metrics by cluster_id via the series API and returns the label value
// If start and end are both zero, the last 1 day is queried.
func (s *LokiProxyService) fetchMetricsLabel(ctx context.Context, clusterMetricsURL, clusterID, metricsName, labelName string, start, end time.Time) (string, error) {
	baseURL := normalizePrometheusURLForLoki(clusterMetricsURL)
	queryURL := strings.TrimRight(baseURL, "/") + "/series"
	match := fmt.Sprintf(`%s{cluster_id="%s"}`, metricsName, clusterID)

	if start.IsZero() && end.IsZero() {
		end = time.Now()
		start = end.Add(-fetchKeyspaceNameDefaultRange)
	}
	if end.Before(start) || end.Equal(start) {
		end = start.Add(fetchKeyspaceNameDefaultRange)
	}
	params := url.Values{}
	params.Add("match[]", match)
	params.Set("start", fmt.Sprintf("%d", start.Unix()))
	params.Set("end", fmt.Sprintf("%d", end.Unix()))
	u, err := url.Parse(queryURL)
	if err != nil {
		return "", fmt.Errorf("parse query URL: %w", err)
	}
	u.RawQuery = params.Encode()
	fullURL := u.String()
	log.Info(ctx, "fetching label from metrics (series)",
		zap.String("clusterID", clusterID),
		zap.String("metricsName", metricsName),
		zap.String("labelName", labelName),
		zap.Time("start", start),
		zap.Time("end", end))
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fullURL, nil)
	if err != nil {
		return "", fmt.Errorf("create request: %w", err)
	}
	resp, err := s.httpClient.Do(req)
	if err != nil {
		return "", fmt.Errorf("request %s: %w", metricsName, err)
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("read response: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("%s returned status %d: %s", metricsName, resp.StatusCode, string(body))
	}
	var result proxyPrometheusSeriesResult
	if err := json.Unmarshal(body, &result); err != nil {
		return "", fmt.Errorf("parse response: %w", err)
	}
	if len(result.Data) == 0 {
		return "", fmt.Errorf("%s returned no series for cluster_id=%s", metricsName, clusterID)
	}
	token, ok := result.Data[0][labelName]
	if !ok || token == "" {
		return "", fmt.Errorf("%s series has no cluster label for cluster_id=%s", metricsName, clusterID)
	}
	return token, nil
}

func normalizePrometheusURLForLoki(rawURL string) string {
	if rawURL == "" {
		return ""
	}
	if !strings.HasPrefix(rawURL, "http://") && !strings.HasPrefix(rawURL, "https://") {
		return "https://" + rawURL
	}
	return rawURL
}

// injectLabelIntoLogQLSelector adds a label to the first stream selector {...} in the LogQL query.
//
// Standard implementation: Grafana Loki provides github.com/grafana/loki/v3/pkg/logql/syntax with
// ParseMatchers(inner, true) ([]*labels.Matcher, error) and MatchersString(matchers) string.
// That parses selector content correctly so quoted strings and regex (e.g. label=~"}{") do not break
// injection. Example:
//
//	inner := content between first '{' and matching '}'
//	matchers, err := syntax.ParseMatchers(inner, true)
//	matchers = append(matchers, labels.MustNewMatcher(labels.MatchEqual, labelName, labelValue))
//	replace selector with "{" + syntax.MatchersString(matchers) + "}"
//
// We use string-based matching here to avoid pulling in github.com/grafana/loki/v3 (large dependency
// tree and possible Go version conflicts). Known limitation: selector content is not parsed; if a
// label value or regex contains '}' the brace matching could be wrong.
func injectLabelIntoLogQLSelector(logql, labelName, labelValue string) string {
	injected, _ := injectLabelIntoLogQLSelectorWithFound(logql, labelName, labelValue)
	return injected
}

func injectLabelIntoLogQLSelectorWithFound(logql, labelName, labelValue string) (string, bool) {
	if labelName == "" || labelValue == "" {
		return logql, false
	}
	idx := strings.Index(logql, "{")
	if idx == -1 {
		return logql, false
	}
	depth := 0
	for i := idx; i < len(logql); i++ {
		switch logql[i] {
		case '{':
			depth++
		case '}':
			depth--
			if depth == 0 {
				insert := fmt.Sprintf(`%s="%s"`, labelName, labelValue)
				inner := strings.TrimSpace(logql[idx+1 : i])
				if inner == "" {
					return logql[:i] + insert + logql[i:], true
				}
				matchers := splitLogQLSelectorMatchers(inner)
				rewritten := make([]string, 0, len(matchers)+1)
				for _, matcher := range matchers {
					matcher = strings.TrimSpace(matcher)
					if matcher == "" || selectorMatcherLabelName(matcher) == labelName {
						continue
					}
					rewritten = append(rewritten, matcher)
				}
				rewritten = append(rewritten, insert)
				return logql[:idx+1] + strings.Join(rewritten, ", ") + logql[i:], true
			}
		}
	}
	return logql, false
}

func splitLogQLSelectorMatchers(inner string) []string {
	var matchers []string
	start := 0
	inQuote := false
	escaped := false
	for i := 0; i < len(inner); i++ {
		ch := inner[i]
		if escaped {
			escaped = false
			continue
		}
		if inQuote && ch == '\\' {
			escaped = true
			continue
		}
		if ch == '"' {
			inQuote = !inQuote
			continue
		}
		if ch == ',' && !inQuote {
			matchers = append(matchers, inner[start:i])
			start = i + 1
		}
	}
	return append(matchers, inner[start:])
}

func selectorMatcherLabelName(matcher string) string {
	matcher = strings.TrimSpace(matcher)
	i := 0
	for i < len(matcher) {
		ch := matcher[i]
		if (ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || ch == '_' {
			i++
			continue
		}
		break
	}
	if i == 0 {
		return ""
	}
	rest := strings.TrimSpace(matcher[i:])
	if strings.HasPrefix(rest, "=") || strings.HasPrefix(rest, "!=") || strings.HasPrefix(rest, "!~") {
		return matcher[:i]
	}
	return ""
}

func injectRequiredLabelIntoLogQLSelector(logql, labelName, labelValue string) (string, bool) {
	return injectLabelIntoLogQLSelectorWithFound(logql, labelName, labelValue)
}

func logQLSelectorWithLabel(labelName, labelValue string) string {
	return fmt.Sprintf(`{%s="%s"}`, labelName, labelValue)
}

// injectLogQLPipelineFilter inserts |= "value" after the first stream selector (first '}').
func injectLogQLPipelineFilter(logql, value string) string {
	if value == "" {
		return logql
	}
	idx := strings.Index(logql, "{")
	if idx == -1 {
		return logql + ` |= "` + value + `"`
	}
	depth := 0
	for i := idx; i < len(logql); i++ {
		if logql[i] == '{' {
			depth++
		} else if logql[i] == '}' {
			depth--
			if depth == 0 {
				filter := ` |= "` + value + `"`
				return logql[:i+1] + filter + logql[i+1:]
			}
		}
	}
	return logql + ` |= "` + value + `"`
}

// namespaceValueInLogQL extracts the value of namespace= from the first stream selector, or empty if not present.
var namespaceInLogQLRE = regexp.MustCompile(`namespace\s*=~?\s*"([^"]*)"`)

func namespaceValueInLogQL(logql string) string {
	m := namespaceInLogQLRE.FindStringSubmatch(logql)
	if len(m) < 2 {
		return ""
	}
	return m[1]
}

// parseLokiTime parses start/end from Loki query params.
// Unix epoch in nanoseconds.
func parseLokiTime(s string) (time.Time, error) {
	if s == "" {
		return time.Time{}, nil
	}
	ts, err := strconv.ParseUint(s, 10, 64)
	if err == nil {
		return time.Unix(0, int64(ts)), nil
	}
	return time.Time{}, fmt.Errorf("cannot parse %q as time", s)
}

// BuildLogQLWithDeployTypeV2 returns the LogQL query with deployTypeV2-specific filters/labels injected.
// For query_range, start and end can be passed so Starter/Essential keyspace_name lookup uses the same range.
func (s *LokiProxyService) BuildLogQLWithDeployTypeV2(ctx context.Context, isQueryParam bool, logql string, info *LokiClusterInfo, start, end time.Time) (string, error) {
	logql = injectLabelIntoLogQLSelector(logql, "cluster_env", s.env)
	deployTypeV2 := strings.ToLower(strings.TrimSpace(info.DeployTypeV2))
	switch {
	case isStarterOrEssential(deployTypeV2):
		o11yCluster, err := s.fetchMetricsLabel(ctx, info.ClusterMetricsURL, info.ClusterID, metricsTidbServerConnections, labelO11yCluster, start, end)
		if err != nil {
			return "", fmt.Errorf("fetch o11y cluster for Starter/Essential: %w", err)
		}
		log.Info(ctx, "injecting cluster for Starter/Essential",
			zap.String("clusterID", info.ClusterID),
			zap.String("cluster", o11yCluster))
		// it's like, freetier/eks/dev-us-east-1-f01
		parts := strings.SplitN(o11yCluster, "/", 3)
		if len(parts) >= 3 {
			o11yCluster = parts[2]
		} else {
			return "", fmt.Errorf("invalid o11y cluster format: %s", o11yCluster)
		}
		return injectLabelIntoLogQLSelector(logql, "cluster", o11yCluster), nil
	case isPremiumOrByoc(deployTypeV2):
		ns := fmt.Sprintf("tidb%s", info.ParentID)
		log.Info(ctx, "injecting namespace for Premium/Byoc",
			zap.String("clusterID", info.ClusterID),
			zap.String("namespace", ns))
		return injectLabelIntoLogQLSelector(logql, "namespace", ns), nil
	case isPremiumOrByocResourcePool(deployTypeV2):
		ns := fmt.Sprintf("tidb%s", info.ClusterID)
		log.Info(ctx, "injecting namespace for Premium/Byoc",
			zap.String("clusterID", info.ClusterID),
			zap.String("namespace", ns))
		return injectLabelIntoLogQLSelector(logql, "namespace", ns), nil
	case deployTypeV2 == deployTypeV2Dedicated:
		nsVal := namespaceValueInLogQL(logql)
		if strings.Contains(nsVal, "dm") || strings.Contains(nsVal, "tidb-admin") {
			log.Info(ctx, "injecting project_id for Dedicated (dm/tidb-admin namespace)",
				zap.String("clusterID", info.ClusterID),
				zap.String("project_id", info.ProjectID))
			return injectLabelIntoLogQLSelector(logql, "project_id", info.ProjectID), nil
		}
		ns := fmt.Sprintf("tidb%s", info.ClusterID)
		log.Info(ctx, "injecting namespace for Dedicated",
			zap.String("clusterID", info.ClusterID),
			zap.String("namespace", ns))
		return injectLabelIntoLogQLSelector(logql, "namespace", ns), nil
	default:
		return logql, nil
	}
}

// IsBYOCLogsCluster reports whether either trusted deploy-type field marks a
// BYOC logical cluster or BYOC resource pool. Inconsistent markers are still
// routed to the BYOC-only path, where validation fails closed.
func IsBYOCLogsCluster(deployType, deployTypeV2 string) bool {
	legacy := strings.ToLower(strings.TrimSpace(deployType))
	v2 := strings.ToLower(strings.TrimSpace(deployTypeV2))
	return legacy == "byoc" || legacy == "byoc-host" ||
		v2 == deployTypeV2Byoc || v2 == deployTypeV2ByocResourcePool
}

type byocLogsScope int

const (
	byocLogsScopeLogical byocLogsScope = iota + 1
	byocLogsScopeResourcePool
)

func validatedBYOCLogsScope(info *LokiClusterInfo) (byocLogsScope, error) {
	if info == nil {
		return 0, fmt.Errorf("BYOC Loki cluster info is required")
	}
	legacy := strings.ToLower(strings.TrimSpace(info.DeployType))
	v2 := strings.ToLower(strings.TrimSpace(info.DeployTypeV2))
	switch {
	case (legacy == "byoc" || legacy == "") && v2 == deployTypeV2Byoc,
		legacy == "byoc" && v2 == "":
		return byocLogsScopeLogical, nil
	case (legacy == "byoc-host" || legacy == "") && v2 == deployTypeV2ByocResourcePool,
		legacy == "byoc-host" && v2 == "":
		return byocLogsScopeResourcePool, nil
	default:
		return 0, fmt.Errorf("inconsistent BYOC logs deploy types")
	}
}

type byocLogQLSelectorRange struct {
	start int
	end   int
}

// singleBYOCLogQLSelectorRange deliberately accepts exactly one stream
// selector. Rewriting only one branch of a multi-selector expression could
// allow a BYOC query to escape its server-controlled namespace.
func singleBYOCLogQLSelectorRange(logql string) (byocLogQLSelectorRange, error) {
	selectors := make([]byocLogQLSelectorRange, 0, 2)
	for i := 0; i < len(logql); {
		switch logql[i] {
		case '#':
			return byocLogQLSelectorRange{}, fmt.Errorf("%w: line comments are not supported", ErrInvalidBYOCLogQuery)
		case '"', '`':
			quote := logql[i]
			i++
			escaped := false
			closed := false
			for i < len(logql) {
				ch := logql[i]
				if quote == '"' && escaped {
					escaped = false
					i++
					continue
				}
				if quote == '"' && ch == '\\' {
					escaped = true
					i++
					continue
				}
				i++
				if ch == quote {
					closed = true
					break
				}
			}
			if !closed {
				return byocLogQLSelectorRange{}, fmt.Errorf("%w: unterminated string", ErrInvalidBYOCLogQuery)
			}
		case '{':
			start := i
			i++
			inQuote := false
			escaped := false
			closed := false
			for i < len(logql) {
				ch := logql[i]
				if escaped {
					escaped = false
					i++
					continue
				}
				if inQuote && ch == '\\' {
					escaped = true
					i++
					continue
				}
				if ch == '"' {
					inQuote = !inQuote
					i++
					continue
				}
				if !inQuote && (ch == '`' || ch == '#') {
					return byocLogQLSelectorRange{}, fmt.Errorf("%w: raw strings and comments are not supported in stream selectors", ErrInvalidBYOCLogQuery)
				}
				if ch == '{' && !inQuote {
					return byocLogQLSelectorRange{}, fmt.Errorf("%w: nested stream selector", ErrInvalidBYOCLogQuery)
				}
				if ch == '}' && !inQuote {
					selectors = append(selectors, byocLogQLSelectorRange{start: start, end: i})
					i++
					closed = true
					break
				}
				i++
			}
			if !closed {
				return byocLogQLSelectorRange{}, fmt.Errorf("%w: unterminated stream selector", ErrInvalidBYOCLogQuery)
			}
			if len(selectors) > 1 {
				return byocLogQLSelectorRange{}, fmt.Errorf("%w: multiple stream selectors are not supported", ErrInvalidBYOCLogQuery)
			}
		default:
			i++
		}
	}
	if len(selectors) != 1 {
		return byocLogQLSelectorRange{}, fmt.Errorf("%w: exactly one stream selector is required", ErrInvalidBYOCLogQuery)
	}
	return selectors[0], nil
}

func injectRequiredBYOCLogQLLabel(logql, labelName, labelValue string) (string, error) {
	if strings.TrimSpace(logql) == "" {
		logql = "{}"
	}
	if strings.TrimSpace(labelName) == "" || strings.TrimSpace(labelValue) == "" {
		return "", fmt.Errorf("required BYOC Loki scope label is empty")
	}
	selector, err := singleBYOCLogQLSelectorRange(logql)
	if err != nil {
		return "", err
	}
	inner := strings.TrimSpace(logql[selector.start+1 : selector.end])
	matchers := splitLogQLSelectorMatchers(inner)
	rewritten := make([]string, 0, len(matchers)+1)
	for _, matcher := range matchers {
		matcher = strings.TrimSpace(matcher)
		if matcher == "" || selectorMatcherLabelName(matcher) == labelName {
			continue
		}
		rewritten = append(rewritten, matcher)
	}
	rewritten = append(rewritten, labelName+"="+strconv.Quote(labelValue))
	return logql[:selector.start+1] + strings.Join(rewritten, ", ") + logql[selector.end:], nil
}

func (s *LokiProxyService) buildBYOCLogQL(logql string, info *LokiClusterInfo) (string, error) {
	scope, err := validatedBYOCLogsScope(info)
	if err != nil {
		return "", err
	}
	if strings.TrimSpace(info.ClusterID) == "" {
		return "", fmt.Errorf("BYOC cluster ID is required")
	}
	env := strings.TrimSpace(s.env)
	if env == "" {
		return "", fmt.Errorf("BYOC cluster environment is required")
	}
	logql, err = injectRequiredBYOCLogQLLabel(logql, "cluster_env", env)
	if err != nil {
		return "", err
	}

	namespaceID := info.ClusterID
	if scope == byocLogsScopeLogical {
		namespaceID = strings.TrimSpace(info.ParentID)
		if namespaceID == "" {
			return "", fmt.Errorf("BYOC parent ID is required")
		}
	}
	return injectRequiredBYOCLogQLLabel(logql, "namespace", "tidb"+namespaceID)
}

func parseBYOCLokiTargetURL(rawURL, env string) (*url.URL, error) {
	if rawURL == "" || strings.TrimSpace(rawURL) != rawURL {
		return nil, fmt.Errorf("BYOC Loki URL must be an absolute HTTP(S) URL")
	}
	parsed, err := url.Parse(rawURL)
	if err != nil || parsed.Opaque != "" || parsed.Host == "" || parsed.Hostname() == "" {
		return nil, fmt.Errorf("BYOC Loki URL must be an absolute HTTP(S) URL")
	}
	scheme := strings.ToLower(parsed.Scheme)
	if scheme != "http" && scheme != "https" {
		return nil, fmt.Errorf("BYOC Loki URL must be an absolute HTTP(S) URL")
	}
	if strings.EqualFold(strings.TrimSpace(env), "prod") && scheme != "https" {
		return nil, fmt.Errorf("production BYOC Loki URL must use HTTPS")
	}
	if parsed.User != nil || parsed.RawQuery != "" || parsed.ForceQuery || parsed.Fragment != "" || parsed.RawFragment != "" {
		return nil, fmt.Errorf("BYOC Loki URL must not contain userinfo, query, or fragment")
	}
	parsed.Scheme = scheme
	return parsed, nil
}

func allowedBYOCLokiRequestHeader(key string) bool {
	switch strings.ToLower(strings.TrimSpace(key)) {
	case "accept", "accept-encoding", "content-type", "user-agent",
		"x-request-id", "x-trace-id", "traceparent", "tracestate":
		return true
	default:
		return false
	}
}

func allowedBYOCLokiResponseHeader(key string) bool {
	switch strings.ToLower(strings.TrimSpace(key)) {
	case "cache-control", "content-encoding", "content-type", "retry-after",
		"x-request-id", "x-ratelimit-limit", "x-ratelimit-remaining", "x-ratelimit-reset":
		return true
	default:
		return false
	}
}

func setBYOCLokiRoutingHeaders(req *http.Request, info *LokiClusterInfo) error {
	tenantID := strings.TrimSpace(info.TenantID)
	provider := strings.TrimSpace(info.Provider)
	region := strings.TrimSpace(info.Region)
	parsedOrgID, err := strconv.ParseUint(tenantID, 10, 64)
	if err != nil || parsedOrgID == 0 || provider == "" || region == "" {
		return fmt.Errorf("BYOC logs routing metadata is incomplete")
	}
	for _, value := range []string{tenantID, provider, region} {
		if strings.ContainsAny(value, "\r\n") {
			return fmt.Errorf("BYOC logs routing metadata is invalid")
		}
	}
	req.Header.Set("X-Biz-Type", "byoc")
	req.Header.Set("X-Org-Id", tenantID)
	req.Header.Set("X-Region-Id", provider+"-"+region)
	return nil
}

func stripBYOCLokiRoutingParams(query url.Values) {
	for _, key := range []string{
		"cluster_id",
		"x-clinic-cluster-id",
		"var-cluster_id",
		"var-tidb_cluster",
		"tidb_cluster",
		"control_plane_info",
		"var-control_plane_info",
		"var-project_id",
	} {
		query.Del(key)
	}
}

var byocLokiLabelName = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]{0,254}$`)

func isBYOCLabelValuesPath(lokiPath string) bool {
	const prefix = "/loki/api/v1/label/"
	const suffix = "/values"
	if !strings.HasPrefix(lokiPath, prefix) || !strings.HasSuffix(lokiPath, suffix) {
		return false
	}
	name := strings.TrimSuffix(strings.TrimPrefix(lokiPath, prefix), suffix)
	return byocLokiLabelName.MatchString(name)
}

// ProxyBYOCRequest is isolated from ProxyRequest so Dedicated, Premium,
// Starter, Essential, System Logs, and serverless-parent logs retain their
// established request rewriting, headers, redirects, and error behavior.
func (s *LokiProxyService) ProxyBYOCRequest(ctx context.Context, lokiPath string, originalURL *url.URL, method string, _ io.Reader, headers http.Header, info *LokiClusterInfo) (*http.Response, error) {
	if s == nil || s.httpClient == nil || originalURL == nil {
		return nil, fmt.Errorf("BYOC Loki proxy is unavailable")
	}
	if method != http.MethodGet {
		return nil, fmt.Errorf("%w: unsupported request method", ErrInvalidBYOCLogQuery)
	}
	if _, err := validatedBYOCLogsScope(info); err != nil {
		return nil, err
	}
	parsedLokiURL, err := parseBYOCLokiTargetURL(info.LokiURL, s.env)
	if err != nil {
		return nil, err
	}

	q := originalURL.Query()
	stripBYOCLokiRoutingParams(q)
	switch {
	case lokiPath == "/loki/api/v1/query_range":
		limit := s.maxLimit
		if raw := q.Get("limit"); raw != "" {
			if parsed, parseErr := strconv.Atoi(raw); parseErr == nil && parsed >= 0 && parsed <= s.maxLimit {
				limit = parsed
			}
		}
		q.Set("limit", strconv.Itoa(limit))
		queries := q["query"]
		if len(queries) != 1 || strings.TrimSpace(queries[0]) == "" {
			return nil, fmt.Errorf("%w: query_range requires exactly one query parameter", ErrInvalidBYOCLogQuery)
		}
		injected, injectErr := s.buildBYOCLogQL(queries[0], info)
		if injectErr != nil {
			return nil, fmt.Errorf("inject BYOC LogQL query: %w", injectErr)
		}
		q.Set("query", injected)
		q.Del("match[]")
	case lokiPath == "/loki/api/v1/labels" || isBYOCLabelValuesPath(lokiPath):
		queries := q["query"]
		matches := q["match[]"]
		if len(queries) > 1 || len(matches) > 1 || (len(queries) > 0 && len(matches) > 0) {
			return nil, fmt.Errorf("%w: labels operations accept one selector", ErrInvalidBYOCLogQuery)
		}
		selector := ""
		if len(queries) == 1 {
			selector = queries[0]
		} else if len(matches) == 1 {
			selector = matches[0]
		}
		injected, injectErr := s.buildBYOCLogQL(selector, info)
		if injectErr != nil {
			return nil, fmt.Errorf("inject BYOC LogQL selector: %w", injectErr)
		}
		q.Set("query", injected)
		q.Del("match[]")
	case lokiPath == "/loki/api/v1/series":
		matches := q["match[]"]
		q.Del("match[]")
		if len(matches) == 0 {
			matches = []string{""}
		}
		for _, selector := range matches {
			injected, injectErr := s.buildBYOCLogQL(selector, info)
			if injectErr != nil {
				return nil, fmt.Errorf("inject BYOC LogQL selector: %w", injectErr)
			}
			q.Add("match[]", injected)
		}
		q.Del("query")
	default:
		return nil, fmt.Errorf("%w: unsupported Loki proxy path", ErrInvalidBYOCLogQuery)
	}

	targetURL := *parsedLokiURL
	targetURL.Path = strings.TrimSuffix(parsedLokiURL.Path, "/") + lokiPath
	targetURL.RawPath = ""
	targetURL.RawQuery = q.Encode()
	// These routes are GET-only. Never forward a caller body, because an
	// upstream form parser could otherwise treat it as a second, unscoped query.
	req, err := http.NewRequestWithContext(ctx, method, targetURL.String(), nil)
	if err != nil {
		return nil, fmt.Errorf("create BYOC Loki request: %w", err)
	}
	for key, values := range headers {
		if !allowedBYOCLokiRequestHeader(key) {
			continue
		}
		for _, value := range values {
			req.Header.Add(key, value)
		}
	}
	if err := setBYOCLokiRoutingHeaders(req, info); err != nil {
		return nil, err
	}
	req.Host = targetURL.Host

	log.Info(ctx, "proxying BYOC logs to o11y",
		zap.String("method", req.Method),
		zap.String("upstreamHost", req.URL.Host),
		zap.String("upstreamPath", req.URL.Path))
	client := *s.httpClient
	client.CheckRedirect = func(*http.Request, []*http.Request) error {
		return http.ErrUseLastResponse
	}
	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("BYOC Loki request failed: %w", err)
	}
	filteredHeaders := make(http.Header)
	for key, values := range resp.Header {
		if !allowedBYOCLokiResponseHeader(key) {
			continue
		}
		for _, value := range values {
			filteredHeaders.Add(key, value)
		}
	}
	resp.Header = filteredHeaders
	if resp.StatusCode != http.StatusOK {
		log.Warn(ctx, "BYOC Loki returned non-200 status", zap.Int("status", resp.StatusCode))
	}
	return resp, nil
}

// ProxyRequest proxies a request to Loki, optionally injecting LogQL labels by deployTypeV2 when info is non-nil.
// For query_range: injects into the "query" parameter. For other endpoints (labels, label/values, series): injects into each "match[]" parameter.
func (s *LokiProxyService) ProxyRequest(ctx context.Context, lokiPath string, originalURL *url.URL, method string, body io.Reader, headers http.Header, info *LokiClusterInfo) (*http.Response, error) {
	// 1. Parse Loki URL
	parsedLokiURL, err := url.Parse(info.LokiURL)
	if err != nil {
		return nil, fmt.Errorf("invalid loki url: %w", err)
	}

	// 2. Build target URL: enforce limit for query_range; optionally rewrite "query" or "match[]" for LogQL injection
	q := originalURL.Query()
	isQueryRange := strings.Contains(lokiPath, "query_range")
	if isQueryRange {
		limitStr := q.Get("limit")
		effectiveLimit := s.maxLimit
		if limitStr != "" {
			if parsed, err := strconv.Atoi(limitStr); err == nil && parsed >= 0 && parsed <= s.maxLimit {
				effectiveLimit = parsed
			}
			// else: invalid or > maxLimit -> keep effectiveLimit = maxLimit
		}
		q.Set("limit", strconv.Itoa(effectiveLimit))
	}
	if isQueryRange {
		var start, end time.Time
		if startStr := q.Get("start"); startStr != "" {
			if t, err := parseLokiTime(startStr); err == nil {
				start = t
			}
		}
		if endStr := q.Get("end"); endStr != "" {
			if t, err := parseLokiTime(endStr); err == nil {
				end = t
			}
		}
		queryParam := q.Get("query")
		injected, err := s.BuildLogQLWithDeployTypeV2(ctx, true, queryParam, info, start, end)
		if err != nil {
			return nil, fmt.Errorf("inject LogQL labels (query): %w", err)
		}
		q.Set("query", injected)
	} else {
		matchValues := q["match[]"]
		if len(matchValues) > 0 {
			var injectedMatch []string
			for _, m := range matchValues {
				injected, err := s.BuildLogQLWithDeployTypeV2(ctx, false, m, info, time.Time{}, time.Time{})
				if err != nil {
					return nil, fmt.Errorf("inject LogQL labels (match[]): %w", err)
				}
				injectedMatch = append(injectedMatch, injected)
			}
			q.Del("match[]")
			for _, v := range injectedMatch {
				q.Add("match[]", v)
			}
		} else {
			injected, err := s.BuildLogQLWithDeployTypeV2(ctx, false, "", info, time.Time{}, time.Time{})
			if err != nil {
				return nil, fmt.Errorf("inject LogQL labels (match[]): %w", err)
			}
			q.Set("match[]", injected)
		}
	}
	rawQuery := q.Encode()

	targetURL := *parsedLokiURL
	targetURL.Path = strings.TrimSuffix(parsedLokiURL.Path, "/") + lokiPath
	targetURL.RawQuery = rawQuery

	// 3. Create and execute request
	req, err := http.NewRequestWithContext(ctx, method, targetURL.String(), body)
	if err != nil {
		return nil, fmt.Errorf("failed to create request: %w", err)
	}

	// Copy headers (except Host)
	for key, values := range headers {
		if strings.ToLower(key) == "host" {
			continue
		}
		for _, value := range values {
			req.Header.Add(key, value)
		}
	}

	// Set Host header to Loki host
	req.Host = targetURL.Host

	log.Info(ctx, "proxying to loki",
		zap.String("method", req.Method),
		zap.String("upstreamHost", req.URL.Host),
		zap.String("upstreamPath", req.URL.Path))

	// Execute request
	resp, err := s.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("loki request failed: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		log.Warn(ctx, "loki returned non-200 status",
			zap.Int("status", resp.StatusCode))
	}

	return resp, nil
}

// ProxySystemRequest proxies a request to a configured control-plane logs endpoint.
// Unlike ProxyRequest, it does not fetch cluster metadata or inject cluster labels.
func (s *LokiProxyService) ProxySystemRequest(ctx context.Context, lokiPath string, originalURL *url.URL, method string, body io.Reader, headers http.Header, lokiURL string) (*http.Response, error) {
	parsedLokiURL, err := url.Parse(lokiURL)
	if err != nil {
		return nil, fmt.Errorf("invalid system logs url: %w", err)
	}
	if parsedLokiURL.Scheme == "" || parsedLokiURL.Host == "" {
		return nil, fmt.Errorf("invalid system logs url: %s", lokiURL)
	}

	q := originalURL.Query()
	if strings.Contains(lokiPath, "query_range") {
		limitStr := q.Get("limit")
		effectiveLimit := s.maxLimit
		if limitStr != "" {
			if parsed, err := strconv.Atoi(limitStr); err == nil && parsed >= 0 && parsed <= s.maxLimit {
				effectiveLimit = parsed
			}
		}
		q.Set("limit", strconv.Itoa(effectiveLimit))
	}

	targetURL := *parsedLokiURL
	targetURL.Path = strings.TrimSuffix(parsedLokiURL.Path, "/") + lokiPath
	targetURL.RawQuery = q.Encode()

	req, err := http.NewRequestWithContext(ctx, method, targetURL.String(), body)
	if err != nil {
		return nil, fmt.Errorf("failed to create request: %w", err)
	}
	for key, values := range headers {
		if strings.ToLower(key) == "host" {
			continue
		}
		for _, value := range values {
			req.Header.Add(key, value)
		}
	}
	req.Host = targetURL.Host

	log.Info(ctx, "proxying to system logs",
		zap.String("method", req.Method),
		zap.String("upstream_host", req.URL.Host),
		zap.String("upstream_path", req.URL.EscapedPath()))

	resp, err := s.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("system logs request failed: %w", err)
	}
	if resp.StatusCode != http.StatusOK {
		log.Warn(ctx, "system logs upstream returned non-200 status",
			zap.Int("status", resp.StatusCode))
	}
	return resp, nil
}

// BuildServerlessParentLogsProxyURL injects the control-plane cluster selector into Loki query params.
func (s *LokiProxyService) BuildServerlessParentLogsProxyURL(lokiPath string, originalURL *url.URL, info ServerlessParentControlPlaneInfo) (*url.URL, error) {
	copied := *originalURL
	q := copied.Query()
	cluster := s.serverlessParentLogsClusterLabel(info)

	if strings.Contains(lokiPath, "query_range") {
		queryParam := q.Get("query")
		if strings.TrimSpace(queryParam) == "" {
			return nil, fmt.Errorf("query parameter is required")
		}
		injected, ok := injectRequiredLabelIntoLogQLSelector(queryParam, "cluster", cluster)
		if !ok {
			return nil, fmt.Errorf("query must contain a LogQL stream selector for cluster injection")
		}
		q.Set("query", injected)
	} else {
		matchValues := q["match[]"]
		q.Del("match[]")
		if len(matchValues) == 0 {
			q.Set("match[]", logQLSelectorWithLabel("cluster", cluster))
		} else {
			for _, match := range matchValues {
				injected, ok := injectRequiredLabelIntoLogQLSelector(match, "cluster", cluster)
				if !ok {
					return nil, fmt.Errorf("match[] must contain a LogQL stream selector for cluster injection")
				}
				q.Add("match[]", injected)
			}
		}
	}

	copied.RawQuery = q.Encode()
	return &copied, nil
}