package cmd

import (
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"text/tabwriter"
	"time"
)

import "github.com/tidbcloud/clinic-cli/internal/clinic"
import "github.com/spf13/cobra"

const (
	lokiDefaultLimit             = 10
	lokiMaxLimit                 = 1000
	lokiGatewayTimeoutMaxRetries = 3
)

func lokiCommandHandler(ctx *RuntimeContext, args []string, stdout io.Writer) error {
	command := newLokiCommand(staticRuntime(ctx), stdout)
	if err := executeCobraCommand(command, args); err != nil {
		return normalizeFamilyCommandError(err, "loki")
	}
	return nil
}

func newLokiCommand(runtime runtimeResolver, stdout io.Writer) *cobra.Command {
	command := newCommand(commandMetadata{
		Use:   "loki",
		Short: "Run Loki log queries and inspect Loki labels.",
		Long: `Run Loki log queries for a cluster.

Subcommands:
  query         Query logs over a time range.
  instance-map   Map Loki streams to node labels for mapping use.
  labels         List available label names.
  label-values   List values for a single label.`,
		Example: `  clinic loki query --cluster-id cluster-1 --start now-5m --end now
  clinic loki instance-map --cluster-id cluster-1 --start now-5m --end now --output json
  clinic loki labels --cluster-id cluster-1`,
	}, stdout)
	command.Args = func(_ *cobra.Command, args []string) error {
		return rejectUnexpectedArgs(args)
	}
	command.RunE = func(c *cobra.Command, _ []string) error {
		return c.Help()
	}
	command.AddCommand(
		newLokiQueryCommand(runtime, stdout),
		newLokiInstanceMapCommand(runtime, stdout),
		newLokiLabelsCommand(runtime, stdout),
		newLokiLabelValuesCommand(runtime, stdout),
	)
	return command
}

type lokiQueryOptions struct {
	ClusterID              string
	Start                  string
	End                    string
	Limit                  int
	Direction              string
	Step                   string
	Output                 string
	Query                  string
	AllowExplicitScope     bool
	SkipNamespaceInjection bool
}

func newLokiQueryCommand(runtime runtimeResolver, stdout io.Writer) *cobra.Command {
	var opts lokiQueryOptions
	command := newCommand(commandMetadata{
		Use:   "query [<logql>] --cluster-id <cluster-id> --start <time> --end <time>",
		Short: "Query logs over a time range.",
		Long: `Run a Loki query_range request.

Behavior:
  --cluster-id can be provided by flag, CLINIC_CLUSTER_ID, or selected profile defaults.
  --start and --end can be provided by flags, CLINIC_RANGE_START/CLINIC_RANGE_END, or selected profile defaults.
  If both are unresolved, the command defaults to --start now-5m and --end now.
  Missing cluster_env and namespace labels are injected when a selector is present, including {}.
  Use --skip-namespace-injection to keep cluster_env injection but skip namespace injection.
  If no LogQL selector is provided, clinic builds a default selector from the selected env and cluster metadata.
  Omitted LogQL and {} remain supported for non-Starter/Essential tiers.
  Starter and Essential require exactly one selector with a non-empty equality cluster="..." matcher.
  Starter and Essential queries route through the regional system-log datasource.
  Essential V2 and Premium routing remains unchanged.

Profile scoping:
  clinic automatically scopes Loki queries to the selected profile/cluster.
  The cluster_id, env, and namespace come from flags, environment variables, or the active profile.

  Do not put cluster_env or namespace in the LogQL selector during normal use.
  clinic injects them automatically. Use {} for content-only searches.

  Explicit cluster_env or namespace selectors are only for intentional raw/cross-profile or cross-namespace investigations.
  Use --allow-explicit-scope to suppress the warning for those advanced investigations.

Time formats:
  --start/--end accept now, now-<duration>, datetime with optional timezone, or epoch in s/ms/ns.`,
		Example: `  clinic loki query --start now-1h --end now '{} |= "Request range exceeds bound"'
  clinic loki query --start now-1h --end now '{} |~ "split|bucket|region"'
  clinic loki query --start now-1h --end now '{app="tidb"} |= "slow query"'`,
	}, stdout)
	command.Args = func(_ *cobra.Command, args []string) error {
		if len(args) > 1 {
			return rejectUnexpectedArgs(args)
		}
		return nil
	}
	command.Flags().StringVar(&opts.ClusterID, "cluster-id", "", "Cluster ID. Or set CLINIC_CLUSTER_ID, or selected profile cluster_id when a profile is active.")
	command.Flags().StringVar(&opts.Start, "start", "", "Range start. Or set CLINIC_RANGE_START, or selected profile range_start when a profile is active.")
	command.Flags().StringVar(&opts.End, "end", "", "Range end. Or set CLINIC_RANGE_END, or selected profile range_end when a profile is active.")
	command.Flags().IntVar(&opts.Limit, "limit", lokiDefaultLimit, "Maximum entries. Default: 10. Range: 1-1000.")
	command.Flags().StringVar(&opts.Direction, "direction", "", "Query direction: forward or backward.")
	command.Flags().StringVar(&opts.Step, "step", "", "Range step.")
	command.Flags().StringVarP(&opts.Output, "output", "o", "", "Output format. Supported: json. Default: log lines.")
	command.Flags().BoolVar(&opts.AllowExplicitScope, "allow-explicit-scope", false, "Suppress warnings for intentional explicit cluster_env/namespace LogQL selectors.")
	command.Flags().BoolVar(&opts.SkipNamespaceInjection, "skip-namespace-injection", false, "Inject cluster_env only and skip namespace injection.")
	command.RunE = func(_ *cobra.Command, args []string) error {
		if len(args) == 1 {
			opts.Query = args[0]
		}
		ctx, err := runtime()
		if err != nil {
			return err
		}
		return runLokiQuery(ctx, opts, stdout)
	}
	return command
}

func runLokiQuery(ctx *RuntimeContext, opts lokiQueryOptions, stdout io.Writer) error {
	resolvedClusterID, err := requiredFlagOrEnvWithContext(ctx, "cluster-id", opts.ClusterID, envClinicClusterID)
	if err != nil {
		return err
	}
	resolvedStart, resolvedEnd, err := resolveRequiredRangeWithContext(ctx, "start", opts.Start, "end", opts.End)
	if err != nil {
		return err
	}
	if opts.Output != "" && opts.Output != "json" {
		return errors.New("output must be one of: json")
	}
	startNS, err := parseTimeFlagToUnixNano("start", resolvedStart)
	if err != nil {
		return err
	}
	endNS, err := parseTimeFlagToUnixNano("end", resolvedEnd)
	if err != nil {
		return err
	}

	if opts.Limit <= 0 {
		return errors.New("limit must be greater than 0")
	}
	if opts.Limit > lokiMaxLimit {
		return fmt.Errorf("limit must be less than or equal to %d", lokiMaxLimit)
	}
	if opts.Direction != "" {
		if opts.Direction != "forward" && opts.Direction != "backward" {
			return errors.New("direction must be one of: forward, backward")
		}
	}

	clusterRef, err := resolveLokiQueryClusterReference(ctx, resolvedClusterID)
	if err != nil {
		return err
	}
	regionalRoute := isRegionalServerlessLokiTier(clusterRef.DeployType())
	var regionalSelector validatedRegionalLokiSelector
	if regionalRoute {
		if err := validateRegionalLokiMetadata(clusterRef); err != nil {
			return err
		}
		regionalSelector, err = validateRegionalLokiQuery(opts.Query)
		if err != nil {
			return fmt.Errorf("cluster %s cannot use regional system-log routing: %w", resolvedClusterID, err)
		}
	}

	var resolvedQuery string
	if opts.Query != "" {
		if !opts.AllowExplicitScope {
			if regionalRoute {
				warnForExplicitRegionalLokiScopeMatchers(ctx, regionalSelector.matchers)
			} else {
				warnForExplicitLokiScopeLabels(ctx, opts.Query)
			}
		}
		if regionalRoute {
			resolvedQuery = applyDefaultRegionalLokiSelectorLabels(
				ctx,
				opts.Query,
				regionalSelector,
				!opts.SkipNamespaceInjection,
			)
		} else {
			resolvedQuery, err = applyDefaultLokiSelectorLabelsWithClusterReference(ctx, clusterRef, opts.Query, !opts.SkipNamespaceInjection)
		}
		if err != nil {
			return err
		}
	} else {
		resolvedQuery, err = buildDefaultLokiQueryWithClusterReference(ctx, clusterRef, !opts.SkipNamespaceInjection)
		if err != nil {
			return err
		}
	}

	params := url.Values{}
	params.Set("query", resolvedQuery)
	params.Set("start", startNS)
	params.Set("end", endNS)
	params.Set("limit", strconv.Itoa(opts.Limit))
	addOptionalString(params, "direction", opts.Direction)
	addOptionalString(params, "step", opts.Step)

	req := clinic.Request{
		Method: http.MethodGet,
		Path:   fmt.Sprintf("/data-proxy/loki/%s/api/v1/query_range", resolvedClusterID),
		Query:  params,
	}
	var datasourceUID string
	if regionalRoute {
		datasourceUID, err = resolveSystemLogDatasource(
			ctx,
			resolvedClusterID,
			clusterRef.ClusterProviderName,
			clusterRef.ClusterRegionName,
		)
		if err != nil {
			return err
		}
		req = systemLogQueryRangeRequest(datasourceUID, params)
	}

	if opts.Output == "json" {
		err = executeLokiQueryRangeJSON(ctx, req, stdout)
	} else {
		err = executeLokiQueryRangeLogs(ctx, req, stdout)
	}
	if err != nil && regionalRoute {
		return fmt.Errorf("query system-log datasource %q for cluster %s: %w", datasourceUID, resolvedClusterID, err)
	}
	return err
}

type lokiInstanceMapOptions struct {
	ClusterID string
	Start     string
	End       string
	Limit     int
	Output    string
}

func newLokiInstanceMapCommand(runtime runtimeResolver, stdout io.Writer) *cobra.Command {
	var opts lokiInstanceMapOptions
	command := newCommand(commandMetadata{
		Use:   "instance-map --cluster-id <cluster-id> --start <time> --end <time>",
		Short: "Build a Loki instance-to-node mapping.",
		Long: `Build a Loki instance-to-node mapping from query_range results.

Behavior:
  --cluster-id can be provided by flag, CLINIC_CLUSTER_ID, or selected profile defaults.
  --start and --end can be provided by flags, CLINIC_RANGE_START/CLINIC_RANGE_END, or selected profile defaults.
  If both are unresolved, the command defaults to --start now-5m and --end now.

Time formats:
  --start/--end accept now, now-<duration>, datetime with optional timezone, or epoch in s/ms/ns.`,
		Example: "  clinic loki instance-map --cluster-id cluster-1 --start now-5m --end now --output table",
	}, stdout)
	command.Args = func(_ *cobra.Command, args []string) error {
		return rejectUnexpectedArgs(args)
	}
	command.Flags().StringVar(&opts.ClusterID, "cluster-id", "", "Cluster ID. Or set CLINIC_CLUSTER_ID, or selected profile cluster_id when a profile is active.")
	command.Flags().StringVar(&opts.Start, "start", "", "Range start. Or set CLINIC_RANGE_START, or selected profile range_start when a profile is active.")
	command.Flags().StringVar(&opts.End, "end", "", "Range end. Or set CLINIC_RANGE_END, or selected profile range_end when a profile is active.")
	command.Flags().IntVar(&opts.Limit, "limit", lokiMaxLimit, "Maximum entries. Default: 1000. Range: 1-1000.")
	command.Flags().StringVarP(&opts.Output, "output", "o", "table", "Output format. Supported: table, json. Default: table.")
	command.RunE = func(_ *cobra.Command, _ []string) error {
		ctx, err := runtime()
		if err != nil {
			return err
		}
		return runLokiInstanceMap(ctx, opts, stdout)
	}
	return command
}

func runLokiInstanceMap(ctx *RuntimeContext, opts lokiInstanceMapOptions, stdout io.Writer) error {
	resolvedClusterID, err := requiredFlagOrEnvWithContext(ctx, "cluster-id", opts.ClusterID, envClinicClusterID)
	if err != nil {
		return err
	}
	resolvedStart, resolvedEnd, err := resolveRequiredRangeWithContext(ctx, "start", opts.Start, "end", opts.End)
	if err != nil {
		return err
	}
	if opts.Output != "table" && opts.Output != "json" {
		return errors.New("output must be one of: table, json")
	}
	startNS, err := parseTimeFlagToUnixNano("start", resolvedStart)
	if err != nil {
		return err
	}
	endNS, err := parseTimeFlagToUnixNano("end", resolvedEnd)
	if err != nil {
		return err
	}
	if opts.Limit <= 0 {
		return errors.New("limit must be greater than 0")
	}
	if opts.Limit > lokiMaxLimit {
		return fmt.Errorf("limit must be less than or equal to %d", lokiMaxLimit)
	}

	resolvedQuery, err := buildDefaultLokiQuery(ctx, resolvedClusterID, true)
	if err != nil {
		return err
	}

	params := url.Values{}
	params.Set("query", resolvedQuery)
	params.Set("start", startNS)
	params.Set("end", endNS)
	params.Set("limit", strconv.Itoa(opts.Limit))

	req := clinic.Request{
		Method: http.MethodGet,
		Path:   fmt.Sprintf("/data-proxy/loki/%s/api/v1/query_range", resolvedClusterID),
		Query:  params,
	}
	return executeLokiInstanceMap(ctx, req, opts.Output, stdout)
}

func executeLokiInstanceMap(ctx *RuntimeContext, req clinic.Request, outputFormat string, stdout io.Writer) error {
	body, err := executeLokiQueryRangeRawWithRetry(ctx, req)
	if err != nil {
		return err
	}

	rows, err := parseLokiInstanceMapRows(body)
	if err != nil {
		return err
	}

	switch outputFormat {
	case "json":
		return writeLokiInstanceMapJSON(rows, stdout)
	case "table":
		return writeLokiInstanceMapTable(rows, stdout)
	default:
		return errors.New("output must be one of: table, json")
	}
}

func buildDefaultLokiQuery(ctx *RuntimeContext, clusterID string, includeNamespace bool) (string, error) {
	clusterRef, err := resolveClusterReferenceByIDWithMetadata(ctx, clusterID)
	if err != nil {
		return "", err
	}
	return buildDefaultLokiQueryWithClusterReference(ctx, clusterRef, includeNamespace)
}

func buildDefaultLokiQueryWithClusterReference(ctx *RuntimeContext, clusterRef clusterReference, includeNamespace bool) (string, error) {
	clusterEnv := buildDefaultLokiClusterEnv(ctx)
	if !includeNamespace {
		return fmt.Sprintf(`{cluster_env=%q}`, clusterEnv), nil
	}

	namespace, err := inferLokiNamespace(clusterRef)
	if err != nil {
		return "", err
	}
	return fmt.Sprintf(`{cluster_env=%q, namespace=%q}`, clusterEnv, namespace), nil
}

func buildDefaultLokiClusterEnv(ctx *RuntimeContext) string {
	clusterEnv := "prod"
	switch {
	case ctx != nil && ctx.SelectedEnv != "":
		clusterEnv = ctx.SelectedEnv
	case ctx != nil && ctx.Config != nil && ctx.Config.ActiveEnv != "":
		clusterEnv = ctx.Config.ActiveEnv
	}
	return clusterEnv
}

func buildDefaultLokiNamespace(ctx *RuntimeContext, clusterID string) (string, error) {
	clusterRef, err := resolveClusterReferenceByIDWithMetadata(ctx, clusterID)
	if err != nil {
		return "", err
	}

	namespace, err := inferLokiNamespace(clusterRef)
	if err != nil {
		return "", err
	}

	return namespace, nil
}

func applyDefaultLokiSelectorLabels(ctx *RuntimeContext, clusterID, query string, includeNamespace bool) (string, error) {
	clusterRef, err := resolveClusterReferenceByIDWithMetadata(ctx, clusterID)
	if err != nil {
		return "", err
	}
	return applyDefaultLokiSelectorLabelsWithClusterReference(ctx, clusterRef, query, includeNamespace)
}

func applyDefaultLokiSelectorLabelsWithClusterReference(ctx *RuntimeContext, clusterRef clusterReference, query string, includeNamespace bool) (string, error) {
	namespace := ""
	if includeNamespace {
		var err error
		namespace, err = inferLokiNamespace(clusterRef)
		if err != nil {
			return "", err
		}
	}
	return applyDefaultLokiSelectorLabelsWithResolvedNamespace(ctx, query, includeNamespace, namespace), nil
}

func applyDefaultLokiSelectorLabelsWithResolvedNamespace(ctx *RuntimeContext, query string, includeNamespace bool, namespace string) string {
	start, end, selectorBody, ok := locateFirstLokiSelector(query)
	if !ok {
		return query
	}

	hasClusterEnv := selectorHasLabelMatcher(selectorBody, "cluster_env")
	hasNamespace := selectorHasLabelMatcher(selectorBody, "namespace")
	missingClusterEnv := !hasClusterEnv
	missingNamespace := includeNamespace && !hasNamespace
	if !missingClusterEnv && !missingNamespace {
		return query
	}

	missingLabels := make([]string, 0, 2)
	if missingClusterEnv {
		clusterEnv := buildDefaultLokiClusterEnv(ctx)
		missingLabels = append(missingLabels, fmt.Sprintf(`cluster_env=%q`, clusterEnv))
	}
	if missingNamespace {
		missingLabels = append(missingLabels, fmt.Sprintf(`namespace=%q`, namespace))
	}

	updatedSelector := strings.TrimSpace(selectorBody)
	if updatedSelector == "" {
		updatedSelector = strings.Join(missingLabels, ", ")
	} else {
		updatedSelector += ", " + strings.Join(missingLabels, ", ")
	}

	return query[:start+1] + updatedSelector + query[end:]
}

func locateFirstLokiSelector(query string) (start int, end int, selectorBody string, ok bool) {
	start = strings.Index(query, "{")
	if start < 0 {
		return 0, 0, "", false
	}

	inQuotes := false
	escaped := false
	for i := start + 1; i < len(query); i++ {
		ch := query[i]
		if inQuotes {
			if escaped {
				escaped = false
				continue
			}
			if ch == '\\' {
				escaped = true
				continue
			}
			if ch == '"' {
				inQuotes = false
			}
			continue
		}

		switch ch {
		case '"':
			inQuotes = true
		case '}':
			return start, i, query[start+1 : i], true
		}
	}

	return 0, 0, "", false
}

func selectorHasLabelMatcher(selectorBody, label string) bool {
	for _, matcher := range splitLokiLabelMatchers(selectorBody) {
		matcher = strings.TrimSpace(matcher)
		if matcher == "" {
			continue
		}

		matcherLabel, _, _, ok := parseLokiLabelMatcher(matcher)
		if ok && matcherLabel == label {
			return true
		}
	}

	return false
}

func splitLokiLabelMatchers(selectorBody string) []string {
	matchers := []string{}
	start := 0
	inQuotes := false
	escaped := false

	for i := 0; i < len(selectorBody); i++ {
		ch := selectorBody[i]
		if inQuotes {
			if escaped {
				escaped = false
				continue
			}
			if ch == '\\' {
				escaped = true
				continue
			}
			if ch == '"' {
				inQuotes = false
			}
			continue
		}

		switch ch {
		case '"':
			inQuotes = true
		case ',':
			matchers = append(matchers, selectorBody[start:i])
			start = i + 1
		}
	}

	matchers = append(matchers, selectorBody[start:])
	return matchers
}

func parseLokiMatcherLabel(matcher string) (string, bool) {
	label, _, _, ok := parseLokiLabelMatcher(matcher)
	return label, ok
}

func parseLokiLabelMatcher(matcher string) (label string, operator string, value string, ok bool) {
	for _, operator := range []string{"=~", "!~", "!=", "="} {
		index := strings.Index(matcher, operator)
		if index <= 0 {
			continue
		}

		label := strings.TrimSpace(matcher[:index])
		if label == "" {
			return "", "", "", false
		}

		return label, operator, strings.TrimSpace(matcher[index+len(operator):]), true
	}

	return "", "", "", false
}

func parseRegionalLokiLabelMatcher(matcher string) (label string, operator string, value string, ok bool) {
	operatorIndex := -1
	matchedOperator := ""
	for _, operator := range []string{"=~", "!~", "!=", "="} {
		index := strings.Index(matcher, operator)
		if index <= 0 {
			continue
		}
		if operatorIndex < 0 || index < operatorIndex || (index == operatorIndex && len(operator) > len(matchedOperator)) {
			operatorIndex = index
			matchedOperator = operator
		}
	}
	if operatorIndex <= 0 {
		return "", "", "", false
	}

	label = strings.TrimSpace(matcher[:operatorIndex])
	if label == "" {
		return "", "", "", false
	}
	return label, matchedOperator, strings.TrimSpace(matcher[operatorIndex+len(matchedOperator):]), true
}

type explicitLokiScopeLabels struct {
	Labels            []string
	NamespaceWildcard bool
}

func warnForExplicitLokiScopeLabels(ctx *RuntimeContext, query string) {
	warnForExplicitLokiScope(ctx, inspectExplicitLokiScopeLabels(query))
}

func warnForExplicitRegionalLokiScopeMatchers(ctx *RuntimeContext, matchers []string) {
	explicit := inspectExplicitRegionalLokiScopeMatchers(matchers)
	if len(explicit.Labels) == 0 {
		return
	}

	message := fmt.Sprintf(
		`explicit regional Loki default scope label(s): %s. Keep the exact physical cluster="..." matcher and omit explicit cluster_env/namespace matchers to use clinic defaults. Use --allow-explicit-scope only for intentional explicit environment or namespace scope.`,
		strings.Join(explicit.Labels, ", "),
	)
	if explicit.NamespaceWildcard {
		message = `namespace=~".*" makes this a cross-namespace Loki search. ` + message
	}
	warningf(ctx, "%s", message)
}

func warnForExplicitLokiScope(ctx *RuntimeContext, explicit explicitLokiScopeLabels) {
	if len(explicit.Labels) == 0 {
		return
	}

	message := fmt.Sprintf(
		"explicit Loki scope label(s): %s. clinic automatically scopes Loki queries from --cluster-id, --env, and the active profile; use {} for normal content-only searches. Use --allow-explicit-scope only for intentional raw/cross-profile or cross-namespace investigations.",
		strings.Join(explicit.Labels, ", "),
	)
	if explicit.NamespaceWildcard {
		message = `namespace=~".*" makes this a cross-namespace Loki search. ` + message
	}
	warningf(ctx, "%s", message)
}

func inspectExplicitLokiScopeLabels(query string) explicitLokiScopeLabels {
	_, _, selectorBody, ok := locateFirstLokiSelector(query)
	if !ok {
		return explicitLokiScopeLabels{}
	}
	return inspectExplicitLokiScopeMatchers(splitLokiLabelMatchers(selectorBody))
}

func inspectExplicitLokiScopeMatchers(matchers []string) explicitLokiScopeLabels {
	return inspectExplicitLokiScopeMatchersWithParser(matchers, parseLokiLabelMatcher)
}

func inspectExplicitRegionalLokiScopeMatchers(matchers []string) explicitLokiScopeLabels {
	return inspectExplicitLokiScopeMatchersWithParser(matchers, parseRegionalLokiLabelMatcher)
}

func inspectExplicitLokiScopeMatchersWithParser(
	matchers []string,
	parseMatcher func(string) (label string, operator string, value string, ok bool),
) explicitLokiScopeLabels {
	seen := make(map[string]struct{}, 2)
	labels := make([]string, 0, 2)
	namespaceWildcard := false
	for _, matcher := range matchers {
		label, operator, value, ok := parseMatcher(strings.TrimSpace(matcher))
		if !ok {
			continue
		}
		switch label {
		case "cluster_env", "namespace":
			if _, exists := seen[label]; !exists {
				seen[label] = struct{}{}
				labels = append(labels, label)
			}
			if label == "namespace" && operator == "=~" && unquoteLokiMatcherValue(value) == ".*" {
				namespaceWildcard = true
			}
		}
	}

	return explicitLokiScopeLabels{
		Labels:            labels,
		NamespaceWildcard: namespaceWildcard,
	}
}

func unquoteLokiMatcherValue(value string) string {
	unquoted, err := strconv.Unquote(strings.TrimSpace(value))
	if err == nil {
		return unquoted
	}
	return strings.Trim(strings.TrimSpace(value), `"`)
}

func inferLokiNamespace(clusterRef clusterReference) (string, error) {
	namespaceClusterID := clusterRef.ClusterID
	switch deployType := clusterRef.DeployType(); deployType {
	case "premium", "byoc":
		if clusterRef.ParentID == "" || clusterRef.ParentID == "null" {
			return "", fmt.Errorf("%s cluster has no parentID: %s", deployType, clusterRef.ClusterID)
		}
		namespaceClusterID = clusterRef.ParentID
	}

	if namespaceClusterID == "" {
		return "", errors.New("cluster has no clusterID")
	}

	return "tidb" + namespaceClusterID, nil
}

type lokiLabelsOptions struct {
	ClusterID string
	Start     string
	End       string
}

func newLokiLabelsCommand(runtime runtimeResolver, stdout io.Writer) *cobra.Command {
	var opts lokiLabelsOptions
	command := newCommand(commandMetadata{
		Use:   "labels --cluster-id <cluster-id>",
		Short: "List Loki label names.",
		Long: `List Loki label names.

Behavior:
  --cluster-id can be provided by flag, CLINIC_CLUSTER_ID, or selected profile defaults.
  --start and --end are optional and default from CLINIC_RANGE_START/CLINIC_RANGE_END or selected profile range defaults.
  If both are unresolved, the request is sent without time bounds.

Time formats:
  --start/--end accept now, now-<duration>, datetime with optional timezone that defaults to local, or epoch in s/ms/ns.`,
	}, stdout)
	command.Args = func(_ *cobra.Command, args []string) error {
		return rejectUnexpectedArgs(args)
	}
	command.Flags().StringVar(&opts.ClusterID, "cluster-id", "", "Cluster ID. Or set CLINIC_CLUSTER_ID, or selected profile cluster_id when a profile is active.")
	command.Flags().StringVar(&opts.Start, "start", "", "Range start. Or set CLINIC_RANGE_START, or selected profile range_start when a profile is active.")
	command.Flags().StringVar(&opts.End, "end", "", "Range end. Or set CLINIC_RANGE_END, or selected profile range_end when a profile is active.")
	command.RunE = func(_ *cobra.Command, _ []string) error {
		ctx, err := runtime()
		if err != nil {
			return err
		}
		return runLokiLabels(ctx, opts, stdout)
	}
	return command
}

func runLokiLabels(ctx *RuntimeContext, opts lokiLabelsOptions, stdout io.Writer) error {
	resolvedClusterID, err := requiredFlagOrEnvWithContext(ctx, "cluster-id", opts.ClusterID, envClinicClusterID)
	if err != nil {
		return err
	}

	params := url.Values{}
	resolvedStart := optionalFlagOrEnvWithContext(ctx, opts.Start, envClinicRangeStart)
	startNS, err := parseOptionalTimeFlagToUnixNano("start", resolvedStart)
	if err != nil {
		return err
	}
	resolvedEnd := optionalFlagOrEnvWithContext(ctx, opts.End, envClinicRangeEnd)
	endNS, err := parseOptionalTimeFlagToUnixNano("end", resolvedEnd)
	if err != nil {
		return err
	}
	addOptionalString(params, "start", startNS)
	addOptionalString(params, "end", endNS)

	return executeClinicJSON(ctx, clinic.Request{
		Method: http.MethodGet,
		Path:   fmt.Sprintf("/data-proxy/loki/%s/api/v1/labels", resolvedClusterID),
		Query:  params,
	}, stdout)
}

type lokiLabelValuesOptions struct {
	ClusterID string
	Label     string
	Start     string
	End       string
}

func newLokiLabelValuesCommand(runtime runtimeResolver, stdout io.Writer) *cobra.Command {
	var opts lokiLabelValuesOptions
	command := newCommand(commandMetadata{
		Use:   "label-values --cluster-id <cluster-id> --label <label>",
		Short: "List values for one Loki label.",
		Long: `List values for one Loki label.

Behavior:
  The label is inserted into the request path and escaped with url.PathEscape.
  --cluster-id can be provided by flag, CLINIC_CLUSTER_ID, or selected profile defaults.
  --start and --end are optional and default from CLINIC_RANGE_START/CLINIC_RANGE_END or selected profile range defaults.

Time formats:
  --start/--end accept now, now-<duration>, datetime with optional timezone that defaults to local, or epoch in s/ms/ns.`,
	}, stdout)
	command.Args = func(_ *cobra.Command, args []string) error {
		return rejectUnexpectedArgs(args)
	}
	command.Flags().StringVar(&opts.ClusterID, "cluster-id", "", "Cluster ID. Or set CLINIC_CLUSTER_ID, or selected profile cluster_id when a profile is active.")
	command.Flags().StringVar(&opts.Label, "label", "", "Label name.")
	command.Flags().StringVar(&opts.Start, "start", "", "Range start. Or set CLINIC_RANGE_START, or selected profile range_start when a profile is active.")
	command.Flags().StringVar(&opts.End, "end", "", "Range end. Or set CLINIC_RANGE_END, or selected profile range_end when a profile is active.")
	command.RunE = func(_ *cobra.Command, _ []string) error {
		ctx, err := runtime()
		if err != nil {
			return err
		}
		return runLokiLabelValues(ctx, opts, stdout)
	}
	return command
}

func runLokiLabelValues(ctx *RuntimeContext, opts lokiLabelValuesOptions, stdout io.Writer) error {
	resolvedClusterID, err := requiredFlagOrEnvWithContext(ctx, "cluster-id", opts.ClusterID, envClinicClusterID)
	if err != nil {
		return err
	}
	if opts.Label == "" {
		return errors.New("missing required --label")
	}

	params := url.Values{}
	resolvedStart := optionalFlagOrEnvWithContext(ctx, opts.Start, envClinicRangeStart)
	startNS, err := parseOptionalTimeFlagToUnixNano("start", resolvedStart)
	if err != nil {
		return err
	}
	resolvedEnd := optionalFlagOrEnvWithContext(ctx, opts.End, envClinicRangeEnd)
	endNS, err := parseOptionalTimeFlagToUnixNano("end", resolvedEnd)
	if err != nil {
		return err
	}
	addOptionalString(params, "start", startNS)
	addOptionalString(params, "end", endNS)

	return executeClinicJSON(ctx, clinic.Request{
		Method: http.MethodGet,
		Path:   fmt.Sprintf("/data-proxy/loki/%s/api/v1/label/%s/values", resolvedClusterID, url.PathEscape(opts.Label)),
		Query:  params,
	}, stdout)
}

func parseTimeFlagToUnixNano(flagName, value string) (string, error) {
	parsed, err := parseFlexibleTimeToUnixNano(value, time.Local)
	if err != nil {
		return "", fmt.Errorf("invalid --%s: %w", flagName, err)
	}
	return strconv.FormatInt(parsed, 10), nil
}

func parseOptionalTimeFlagToUnixNano(flagName, value string) (string, error) {
	if value == "" {
		return "", nil
	}
	return parseTimeFlagToUnixNano(flagName, value)
}

func executeLokiQueryRangeLogs(ctx *RuntimeContext, req clinic.Request, stdout io.Writer) error {
	body, err := executeLokiQueryRangeRawWithRetry(ctx, req)
	if err != nil {
		return err
	}

	return writeLokiLogLines(stdout, body)
}

func executeLokiQueryRangeJSON(ctx *RuntimeContext, req clinic.Request, stdout io.Writer) error {
	body, err := executeLokiQueryRangeRawWithRetry(ctx, req)
	if err != nil {
		return err
	}

	return writeFormattedJSONOrRaw(stdout, body)
}

func executeLokiQueryRangeRawWithRetry(ctx *RuntimeContext, req clinic.Request) ([]byte, error) {
	maxAttempts := lokiGatewayTimeoutMaxRetries + 1
	for attempt := 1; attempt <= maxAttempts; attempt++ {
		var body []byte
		if err := doClinicRequest(ctx, req, &body); err != nil {
			if !isGatewayTimeoutError(err) || attempt == maxAttempts {
				return nil, err
			}

			diagnosticf(ctx, "loki query received 504 Gateway Timeout, retrying attempt %d/%d", attempt+1, maxAttempts)
			continue
		}

		return body, nil
	}

	return nil, errors.New("unreachable loki query retry loop")
}

func isGatewayTimeoutError(err error) bool {
	var httpErr *clinic.HTTPError
	return errors.As(err, &httpErr) && httpErr.StatusCode == http.StatusGatewayTimeout
}

type lokiQueryRangeResponse struct {
	Data lokiQueryRangeData `json:"data"`
}

type lokiQueryRangeData struct {
	ResultType string                 `json:"resultType"`
	Result     []lokiQueryRangeStream `json:"result"`
}

type lokiQueryRangeStream struct {
	Stream map[string]string `json:"stream"`
	Values [][]string        `json:"values"`
}

type lokiInstanceMapRow struct {
	Node               string `json:"node"`
	DataServerInstance string `json:"data_server_instance"`
	LokiInstance       string `json:"loki_instance"`
	IP                 string `json:"ip"`
}

var lokiNodeIPPattern = regexp.MustCompile(`(?:\d{1,3}\.){3}\d{1,3}$`)

func parseLokiInstanceMapRows(body []byte) ([]lokiInstanceMapRow, error) {
	var response lokiQueryRangeResponse
	if err := json.Unmarshal(body, &response); err != nil {
		return nil, fmt.Errorf("decode loki response: %w", err)
	}

	if response.Data.ResultType != "" && response.Data.ResultType != "streams" {
		return nil, fmt.Errorf("unexpected loki result type: %s", response.Data.ResultType)
	}

	rows := make(map[string]lokiInstanceMapRow, len(response.Data.Result))
	for _, stream := range response.Data.Result {
		node, lokiInstance := strings.TrimSpace(stream.Stream["node"]), strings.TrimSpace(stream.Stream["instance"])
		if node == "" || lokiInstance == "" {
			continue
		}

		row := lokiInstanceMapRow{
			Node:               node,
			DataServerInstance: node,
			LokiInstance:       lokiInstance,
			IP:                 deriveIPFromNode(node),
		}
		rows[row.Node+"\x00"+row.LokiInstance] = row
	}

	mapped := make([]lokiInstanceMapRow, 0, len(rows))
	for _, row := range rows {
		mapped = append(mapped, row)
	}
	sort.Slice(mapped, func(i, j int) bool {
		a, b := mapped[i], mapped[j]
		if a.Node != b.Node {
			return a.Node < b.Node
		}
		if a.DataServerInstance != b.DataServerInstance {
			return a.DataServerInstance < b.DataServerInstance
		}
		if a.LokiInstance != b.LokiInstance {
			return a.LokiInstance < b.LokiInstance
		}
		return a.IP < b.IP
	})

	return mapped, nil
}

func deriveIPFromNode(node string) string {
	ipCandidate := lokiNodeIPPattern.FindString(node)
	if ipCandidate == "" {
		return ""
	}

	ip := net.ParseIP(ipCandidate)
	if ip == nil {
		return ""
	}
	return ip.To4().String()
}

func writeLokiInstanceMapJSON(rows []lokiInstanceMapRow, stdout io.Writer) error {
	body, err := json.Marshal(rows)
	if err != nil {
		return fmt.Errorf("marshal loki instance map rows: %w", err)
	}
	return writeFormattedJSONOrRaw(stdout, body)
}

func writeLokiInstanceMapTable(rows []lokiInstanceMapRow, stdout io.Writer) error {
	writer := tabwriter.NewWriter(stdout, 0, 0, 2, ' ', 0)
	_, err := io.WriteString(writer, "NODE\tDATA_SERVER_INSTANCE\tLOKI_INSTANCE\tIP\t\n")
	if err != nil {
		return err
	}

	for _, row := range rows {
		line := fmt.Sprintf("%s\t%s\t%s\t%s\t\n", row.Node, row.DataServerInstance, row.LokiInstance, row.IP)
		if _, err := io.WriteString(writer, line); err != nil {
			return err
		}
	}

	return writer.Flush()
}

func writeLokiLogLines(stdout io.Writer, body []byte) error {
	var response lokiQueryRangeResponse
	if err := json.Unmarshal(body, &response); err != nil {
		return fmt.Errorf("decode loki response: %w", err)
	}

	if response.Data.ResultType != "" && response.Data.ResultType != "streams" {
		return fmt.Errorf("unexpected loki result type: %s", response.Data.ResultType)
	}

	for _, stream := range response.Data.Result {
		prefix := formatLokiStreamPrefix(stream.Stream)
		for _, value := range stream.Values {
			if len(value) < 2 {
				continue
			}
			logLine := strings.TrimRight(value[1], "\r\n")
			if _, err := io.WriteString(stdout, prefix+logLine+"\n"); err != nil {
				return err
			}
		}
	}

	return nil
}

func formatLokiStreamPrefix(streamLabels map[string]string) string {
	instance := "unknown"
	if streamLabels != nil {
		candidate := strings.TrimSpace(streamLabels["instance"])
		if candidate != "" {
			instance = candidate
		}
	}

	return "[instance=" + instance + "] "
}