package service

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

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/retry"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/cluster-service-ng/internal/cluster-agent/internal/sqlexec"
	"github.com/tidbcloud/cluster-service-ng/internal/cluster-agent/internal/utils"
)

const tlsClientSecret = "db-cluster-client-secret"

type StandbyTiDBStatus struct {
	State        string `json:"state"`
	KeyspaceName string `json:"keyspace_name"`
}

// decryptPassword validates and decrypts the encrypted password bytes.
func (s *AgentService) decryptPassword(encryptedPassword []byte) (string, error) {
	if len(encryptedPassword) == 0 {
		return "", fmt.Errorf("encrypted password is empty")
	}
	password, err := s.encryptor.Decrypt(encryptedPassword)
	if err != nil {
		return "", fmt.Errorf("failed to decrypt the password. ensure the public key is correct: %w", err)
	}
	return string(password), nil
}

// ResetRootPassword handles the ResetRootPasswordReq request.
//
// Deprecated: Use ResetPassword instead. This method delegates to ResetPassword with target=ROOT.
func (s *AgentService) ResetRootPassword(
	ctx context.Context, req *apiv1.ResetRootPasswordReq) (*apiv1.ResetRootPasswordResp, error) {
	resp, err := s.ResetPassword(ctx, &apiv1.ResetPasswordReq{
		ClusterId:         req.GetClusterId(),
		EncryptedPassword: req.GetEncryptedPassword(),
		Target:            apiv1.PasswordResetTarget_ROOT,
	})
	if err != nil {
		return &apiv1.ResetRootPasswordResp{Message: resp.GetMessage()}, err
	}
	return &apiv1.ResetRootPasswordResp{Message: resp.GetMessage()}, nil
}

// targetToUsername maps a PasswordResetTarget enum to the corresponding database username.
// UNSPECIFIED defaults to "root" for backward compatibility with old ResetRootPasswordReq.
func targetToUsername(target apiv1.PasswordResetTarget) (string, error) {
	switch target {
	case apiv1.PasswordResetTarget_ROOT, apiv1.PasswordResetTarget_PASSWORD_RESET_TARGET_UNSPECIFIED:
		return "root", nil
	case apiv1.PasswordResetTarget_CLOUD_ADMIN:
		return "cloud_admin", nil
	case apiv1.PasswordResetTarget_PLATFORM_ADMIN:
		return "platform_admin", nil
	case apiv1.PasswordResetTarget_MAINTENANCE_ADMIN:
		return "maintenance_admin", nil
	default:
		return "", status.Errorf(codes.InvalidArgument, "unsupported password reset target: %v", target)
	}
}

func logResetPasswordViaPrivilegedTiDBError(logger sqlexecErrorLogger, err error) string {
	return logSQLError(logger, "failed to reset password via privileged TiDB", err)
}

// ResetPassword resets the password for a specified database user.
// It tries V2 (TiDB HTTP API) first. If the TiDB version doesn't support the API (404),
// it falls back to V1 (privileged TiDB with ALTER USER), then V0 (direct MySQL connection).
func (s *AgentService) ResetPassword(
	ctx context.Context, req *apiv1.ResetPasswordReq,
) (*apiv1.ResetPasswordResp, error) {
	logger := tilog.WithContextS(ctx)

	password, err := s.decryptPassword(req.GetEncryptedPassword())
	if err != nil {
		msg := fmt.Sprintf("failed to validate the request: %v", err)
		logger.Errorf(msg)
		return &apiv1.ResetPasswordResp{Message: msg}, err
	}

	username, err := targetToUsername(req.GetTarget())
	if err != nil {
		return &apiv1.ResetPasswordResp{}, err
	}
	logger = logger.With("cluster_id", req.GetClusterId(), "target_user", username)

	// V2-first: try HTTP API
	err = s.resetPasswordViaHTTP(ctx, req.GetClusterId(), username, password, req.GetUseTidbWorker())
	if err == nil {
		logger.Infof("password reset via HTTP API succeeded")
		return &apiv1.ResetPasswordResp{Message: "success"}, nil
	}

	// Fallback on:
	// 1) 404 from V2 API (TiDB version doesn't support this API, or user not found), or
	// 2) worker address resolution failure when use_tidb_worker=true (worker group absent).
	var httpErr *TiDBHTTPError
	shouldFallback := errors.As(err, &httpErr) && httpErr.StatusCode == http.StatusNotFound
	if req.GetUseTidbWorker() && errors.Is(err, utils.ErrNoWorkerTiDBGroup) {
		shouldFallback = true
		logger.Warnf("worker TiDB group not found, falling back to privileged TiDB: %v", err)
	}
	if !shouldFallback {
		msg := fmt.Sprintf("failed to reset password via TiDB HTTP API: %v", err)
		logger.Errorf(msg)
		return &apiv1.ResetPasswordResp{Message: msg}, err
	}

	// V1 fallback: use privileged TiDB.
	// When using the TiDB worker (use_tidb_worker=true), the privileged TiDB must be
	// activated with the SYSTEM keyspace.
	keyspace := req.GetClusterId()
	if req.GetUseTidbWorker() {
		keyspace = systemKeyspace
	}
	logger.Warnf("TiDB HTTP API returned 404, falling back to privileged TiDB: %v", err)
	if err := s.resetPasswordViaPrivilegedTiDB(ctx, keyspace, username, password); err != nil {
		v1ErrMsg := logResetPasswordViaPrivilegedTiDBError(logger, err)

		// V0 fallback: direct MySQL connection (final fallback)
		logger.Warnf("privileged TiDB failed, falling back to direct MySQL connection (V0): %s", v1ErrMsg)
		if v0Err := resetPasswordDirectFunc(s, ctx, req.GetClusterId(), username, password, req.GetUseTidbWorker()); v0Err != nil {
			msg := fmt.Sprintf("all password reset methods failed. V1: %s, V0 direct MySQL: %v", v1ErrMsg, v0Err)
			logger.Errorf(msg)
			return &apiv1.ResetPasswordResp{Message: msg},
				fmt.Errorf("%s", msg)
		}
	}

	return &apiv1.ResetPasswordResp{Message: "success"}, nil
}

// resetPasswordViaHTTP resets a user's password via TiDB Internal User Management HTTP API (V2).
func (s *AgentService) resetPasswordViaHTTP(ctx context.Context, clusterID, username, password string, useTiDBWorker bool) error {
	var deps *v2HTTPDeps
	var err error
	if useTiDBWorker {
		deps, err = s.getWorkerV2HTTPDepsResolver()(ctx, s.dataplaneCli)
	} else {
		deps, err = s.getV2HTTPDepsResolver()(ctx, s.dataplaneCli, clusterID)
	}
	if err != nil {
		return fmt.Errorf("failed to resolve V2 HTTP dependencies: %w", err)
	}

	apiURL, err := buildTiDBInternalAPIURL(deps.baseURL, tidbInternalAPIPathResetPassword(username))
	if err != nil {
		return fmt.Errorf("failed to build TiDB internal API URL: %w", err)
	}

	payload, err := json.Marshal(map[string]string{
		"new_password": password,
	})
	if err != nil {
		return fmt.Errorf("failed to marshal request body: %w", err)
	}

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(payload))
	if err != nil {
		return fmt.Errorf("failed to create HTTP request: %w", err)
	}
	httpReq.Header.Set("Content-Type", "application/json")

	return doTiDBHTTPRequest(deps.client, httpReq, http.StatusOK)
}

// workerV2HTTPDepsResolverFunc resolves the HTTP client and base URL for worker TiDB V2 methods.
type workerV2HTTPDepsResolverFunc func(ctx context.Context, cli client.Client) (*v2HTTPDeps, error)

// defaultResolveWorkerV2HTTPDeps is the default implementation for resolving worker TiDB V2 HTTP dependencies.
var defaultResolveWorkerV2HTTPDeps workerV2HTTPDepsResolverFunc = func(ctx context.Context, cli client.Client) (*v2HTTPDeps, error) {
	namespace := os.Getenv("NAMESPACE")
	if namespace == "" {
		return nil, fmt.Errorf("NAMESPACE is empty")
	}

	sqlAddr, err := resolveWorkerTiDBAddr(ctx, cli)
	if err != nil {
		return nil, fmt.Errorf("failed to resolve worker TiDB address: %w", err)
	}

	baseURL, err := sqlAddrToHTTPBaseURL(sqlAddr)
	if err != nil {
		return nil, fmt.Errorf("failed to convert worker TiDB SQL address to HTTP URL: %w", err)
	}

	tlsConfig, err := GetTLSConfig(ctx, cli, namespace, tlsClientSecret)
	if err != nil {
		return nil, fmt.Errorf("failed to get TLS config: %w", err)
	}
	tlsConfig = tlsConfig.Clone()
	tlsConfig.ServerName = fmt.Sprintf("tidb.db-cluster.%s.svc", namespace)

	return &v2HTTPDeps{
		client: &http.Client{
			Timeout: 30 * time.Second,
			Transport: &http.Transport{
				TLSClientConfig:       tlsConfig,
				TLSHandshakeTimeout:   15 * time.Second,
				ResponseHeaderTimeout: 15 * time.Second,
				IdleConnTimeout:       90 * time.Second,
			},
		},
		baseURL: baseURL,
	}, nil
}

// getWorkerV2HTTPDepsResolver returns the worker resolver function, using the injected one if set,
// or falling back to the default implementation.
func (s *AgentService) getWorkerV2HTTPDepsResolver() workerV2HTTPDepsResolverFunc {
	if s.resolveWorkerHTTPDeps != nil {
		return s.resolveWorkerHTTPDeps
	}
	return defaultResolveWorkerV2HTTPDeps
}

// resetPasswordViaPrivilegedTiDB resets a user's password via privileged TiDB (V1).
// This acquires the privileged TiDB mutex to prevent concurrent activation conflicts.
func (s *AgentService) resetPasswordViaPrivilegedTiDB(ctx context.Context, clusterID, username, password string) error {
	s.mutexPrivilegedTiDBActivation.Lock()
	defer s.mutexPrivilegedTiDBActivation.Unlock()

	namespace := os.Getenv("NAMESPACE")
	lifecycle := &privilegedTiDBLifecycle{svc: s, namespace: namespace, enableAuditlog: true, username: "maintenance_admin"}
	executor := newPrivilegedTiDBExecutor(lifecycle)
	cmd := &sqlexec.ResetPasswordCommand{
		Username: username,
		Password: sqlexec.NewSecretString(password),
	}

	_, err := executor.Execute(ctx, clusterID, cmd)
	return err
}

func (s *AgentService) prepareForPrivilegedTiDBExecution(ctx context.Context, clusterID, namespace, tidbAddr string, enableAuditlog bool) error {
	tlsConfig, err := GetTLSConfig(ctx, s.dataplaneCli, namespace, tlsClientSecret)
	if err != nil {
		return fmt.Errorf("failed to get TLS config from secret %s/%s: %w", namespace, tlsClientSecret, err)
	}
	httpClient := http.Client{
		Transport: &http.Transport{
			TLSClientConfig: tlsConfig,
		},
	}

	// Check if the privileged TiDB is available (standby or activated with our
	// own keyspace). If occupied by a different keyspace we fail fast and let
	// the V0 fallback in ResetPassword take over -- the previous request's
	// deactivation goroutine should free the pod within a few seconds (see
	// PrivilegedTiDBDeactivationWait), and queueing here would only consume
	// the request ctx and starve V0.
	logger := tilog.WithContextS(ctx).With("cluster_id", clusterID, "namespace", namespace, "tidb_addr", tidbAddr)
	logger.Infof("preparing privileged TiDB execution")

	// Self-heal: privileged TiDB can get stuck activated for a keyspace whose
	// cluster has been deleted (e.g., the deactivation goroutine was lost to a
	// cluster-agent restart). If we detect that, force-deactivate so the
	// availability check below sees a fresh pod. Errors are non-fatal.
	//
	// Intentionally NO bounded wait after a successful preempt: adding one
	// just rebuilds (in narrower scope) the polling loop we removed for the
	// storm path. The recreate window can land on the immediate availability
	// check; in that case the request falls through to V0, which covers the
	// fresh-init cluster create scenario this preempt path was added to help.
	if preempted, err := s.preemptStalePrivilegedTiDB(ctx, httpClient, tidbAddr, namespace, clusterID); err != nil {
		logger.Warnf("stale-state preemption check failed: %v", err)
	} else if preempted {
		logger.Infof("preempted stale privileged TiDB; relying on availability check + V0 fallback during pod recreate")
	}

	available, err := s.isPrivilegedTiDBAvailable(ctx, httpClient, tidbAddr, clusterID)
	if err != nil {
		return fmt.Errorf("failed to check if the privileged TiDB is available: %w", err)
	} else if !available {
		logger.Warnf("privileged TiDB unavailable, activated for another cluster/keyspace")
		return fmt.Errorf("the privileged TiDB is not available (activated for another cluster/keyspace)")
	}

	if enableAuditlog && clusterID != systemKeyspace {
		// update the auditlog config annotation to match the audit log configuration of the current cluster/keyspace.
		auditConfig, err := s.getTiDBAuditConfig(ctx, namespace, clusterID)
		if err != nil {
			return fmt.Errorf("failed to get the audit log config of the current cluster: %w", err)
		}
		if err := s.updateTiDBAuditConfig(ctx, namespace, auditConfig); err != nil {
			return fmt.Errorf("failed to update the audit log config of the privileged TiDB: %w", err)
		}
	}

	// activate the standby privileged TiDB for the specified cluster/keyspace.
	if err := s.activatedPrivilegedTiDB(ctx, httpClient, tidbAddr, clusterID); err != nil {
		return fmt.Errorf("failed to activate the privileged TiDB: %w", err)
	}
	return nil
}

// fetchPrivilegedTiDBStatus calls /tidb-pool/status and returns the parsed
// state + keyspace binding. It is the shared HTTP helper for availability
// checks and stale-state recovery.
func (s *AgentService) fetchPrivilegedTiDBStatus(ctx context.Context, httpClient http.Client, tidbAddr string) (*StandbyTiDBStatus, error) {
	standbyStatusURL, err := buildPrivilegedTiDBPoolURL(tidbAddr, "/tidb-pool/status")
	if err != nil {
		return nil, fmt.Errorf("failed to build privileged TiDB status URL: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, standbyStatusURL, http.NoBody)
	if err != nil {
		return nil, fmt.Errorf("failed to create request to get privileged TiDB status: %w", err)
	}
	//nolint:gosec // G704: request URL is built by buildPrivilegedTiDBPoolURL with internal host/path validation.
	resp, err := httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("failed to get privileged TiDB status: %w", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()

	if resp.StatusCode != http.StatusOK {
		bodyBytes, readErr := io.ReadAll(io.LimitReader(resp.Body, maxTiDBHTTPErrorBodyBytes))
		body := string(bodyBytes)
		if readErr != nil {
			if body == "" {
				body = fmt.Sprintf("failed to read response body: %v", readErr)
			} else {
				body = fmt.Sprintf("%s (failed to finish reading response body: %v)", body, readErr)
			}
		}
		return nil, fmt.Errorf("privileged TiDB status endpoint returned HTTP %d: %s", resp.StatusCode, body)
	}

	var status StandbyTiDBStatus
	if err := json.NewDecoder(resp.Body).Decode(&status); err != nil {
		return nil, fmt.Errorf("failed to decode privileged TiDB status response: %w", err)
	}
	return &status, nil
}

func (s *AgentService) isPrivilegedTiDBAvailable(ctx context.Context, httpClient http.Client, tidbAddr, clusterID string) (bool, error) {
	logger := tilog.WithContextS(ctx).With("cluster_id", clusterID, "tidb_addr", tidbAddr)
	logger.Infof("checking privileged TiDB availability")

	status, err := s.fetchPrivilegedTiDBStatus(ctx, httpClient, tidbAddr)
	if err != nil {
		return false, err
	}

	logger.Infof("privileged TiDB status: state=%s, keyspace=%s", status.State, status.KeyspaceName)

	if status.State != "activated" || status.KeyspaceName == "" {
		// the privileged TiDB is standby now, so we can activate it for the current cluster/keyspace.
		return true, nil
	} else if status.State == "activated" && status.KeyspaceName == clusterID {
		// the privileged TiDB is already activated for the current cluster/keyspace.
		return true, nil
	}
	// the privileged TiDB is activated for another cluster/keyspace.
	logger.Warnf("privileged TiDB activated for keyspace %q, but need %q", status.KeyspaceName, clusterID)
	return false, nil
}

func (s *AgentService) getTiDBAuditConfig(ctx context.Context, ns, clusterID string) (string, error) {
	// find any normal TiDB Pod in the cluster to get the audit log config
	var podList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &podList,
		client.InNamespace(ns),
		client.MatchingLabels{"pingcap.com/component": "tidb", "tags.tidbcloud.com/cluster": clusterID}); err != nil {
		return "", fmt.Errorf("failed to list pods in namespace %s with cluster ID %s: %w", ns, clusterID, err)
	}
	if len(podList.Items) == 0 {
		return "", fmt.Errorf("no TiDB pods found in namespace %s with cluster ID %s", ns, clusterID)
	}

	pod := podList.Items[0]
	auditConfig, ok := pod.Annotations["tidb.auditlog.config"]
	if !ok {
		return "", fmt.Errorf("audit log config annotation not found in pod %s/%s", pod.Namespace, pod.Name)
	}
	return auditConfig, nil
}

func (s *AgentService) updateTiDBAuditConfig(ctx context.Context, ns, auditConfig string) error {
	// update the `tidb.auditlog.config` annotation of the privileged TiDB Pod
	var podList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &podList,
		client.InNamespace(ns),
		client.MatchingLabels{PingcapComponentLabelKey: PrivilegedTiDBComponentType}); err != nil {
		return fmt.Errorf("failed to list privileged TiDB pods in namespace %s: %w", ns, err)
	}
	// Filter to Running pods -- during deactivation (pod delete/recreate),
	// a Terminating pod and a new pod can coexist briefly.
	pod, err := findSingleRunningPod(podList.Items, ns, PrivilegedTiDBComponentType)
	if err != nil {
		return err
	}
	// for the privileged TiDB, it can be shut down quickly, so we set
	// a shorter gracefulStopWaitSeconds (less that the default 30s for terminationGracePeriodSeconds)
	auditConfig = replaceAuditLogStopWait(auditConfig)

	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
		if err := s.dataplaneCli.Get(ctx, client.ObjectKeyFromObject(pod), pod); err != nil {
			return fmt.Errorf("failed to get latest pod version: %w", err)
		}
		if pod.Annotations == nil {
			pod.Annotations = make(map[string]string)
		}
		pod.Annotations["tidb.auditlog.config"] = auditConfig
		return s.dataplaneCli.Update(ctx, pod)
	}); err != nil {
		return fmt.Errorf("failed to update privileged TiDB pod %s/%s: %w", pod.Namespace, pod.Name, err)
	}
	return nil
}

func (s *AgentService) activatedPrivilegedTiDB(ctx context.Context, httpClient http.Client, tidbAddr, clusterID string) error {
	standbyActivateURL, err := buildPrivilegedTiDBPoolURL(tidbAddr, "/tidb-pool/activate")
	if err != nil {
		return fmt.Errorf("failed to build privileged TiDB activation URL: %w", err)
	}
	payload := map[string]string{
		"keyspace_name": clusterID,
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("failed to marshal activate payload: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, standbyActivateURL, bytes.NewReader(payloadBytes))
	if err != nil {
		return fmt.Errorf("failed to create request to activate privileged TiDB: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	//nolint:gosec // G704: request URL is built by buildPrivilegedTiDBPoolURL with internal host/path validation.
	resp, err := httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("failed to activate privileged TiDB: %w", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()

	if resp.StatusCode != http.StatusOK {
		bodyBytes, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("failed to activate privileged TiDB, status: %s, response: %s", resp.Status, string(bodyBytes))
	}
	return nil
}

// buildPrivilegedTiDBPoolURL constructs the HTTPS URL for the privileged TiDB pool API.
// Variable (not func) to allow test overrides.
var buildPrivilegedTiDBPoolURL = func(host, path string) (string, error) {
	return buildHTTPSURL(host, 10080, path)
}

// deactivatePrivilegedTiDB deletes the privileged TiDB pod to make it standby again.
// If expectedUID is non-empty, only the pod with matching UID will be deleted. This
// prevents racing with a newer pod that may have been activated by another request
// after our original pod was replaced. An empty expectedUID falls back to deleting
// whichever running pod is found (legacy behavior).
func (s *AgentService) deactivatePrivilegedTiDB(ns string, expectedUID types.UID) error {
	// use a new context instead of the original context,
	// so that we can deactivate the privileged TiDB after the request is done
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	var podList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &podList,
		client.InNamespace(ns),
		client.MatchingLabels{PingcapComponentLabelKey: PrivilegedTiDBComponentType}); err != nil {
		return fmt.Errorf("failed to list privileged TiDB pods in namespace %s: %w", ns, err)
	}
	pod, err := findSingleRunningPod(podList.Items, ns, PrivilegedTiDBComponentType)
	if err != nil {
		// No running pod means it was already deleted (by another deactivation or
		// external action); the controller will recreate a standby pod. Nothing to do.
		if errors.Is(err, errNoRunningPods) {
			return nil
		}
		return err
	}
	// If we recorded a UID and the current pod doesn't match, our pod was already
	// replaced (likely deactivated by someone else). Don't delete the new pod --
	// it may have been activated by another request for a different keyspace.
	if expectedUID != "" && pod.UID != expectedUID {
		tilog.WithContextS(ctx).Infow("skipping deactivation: running pod UID does not match activated pod UID; pod may have been replaced",
			zap.String("namespace", ns),
			zap.String("pod_name", pod.Name),
			zap.String("expected_uid", string(expectedUID)),
			zap.String("actual_uid", string(pod.UID)),
		)
		return nil
	}
	if err := s.dataplaneCli.Delete(ctx, pod); err != nil {
		return fmt.Errorf("failed to delete privileged TiDB pod %s/%s in namespace %s: %w", pod.Namespace, pod.Name, ns, err)
	}
	return nil
}

// getPrivilegedTiDBPodUID returns the UID of the currently running privileged TiDB pod
// in the given namespace. Used to record pod identity at activation time so deactivation
// can avoid deleting a replacement pod that another request may have activated.
func (s *AgentService) getPrivilegedTiDBPodUID(ctx context.Context, ns string) (types.UID, error) {
	var podList corev1.PodList
	if err := s.dataplaneCli.List(ctx, &podList,
		client.InNamespace(ns),
		client.MatchingLabels{PingcapComponentLabelKey: PrivilegedTiDBComponentType}); err != nil {
		return "", fmt.Errorf("failed to list privileged TiDB pods in namespace %s: %w", ns, err)
	}
	pod, err := findSingleRunningPod(podList.Items, ns, PrivilegedTiDBComponentType)
	if err != nil {
		return "", err
	}
	return pod.UID, nil
}

// getPrivilegedTiDBServiceAddress returns the base service address of the privileged TiDB service, no port included.
func (s *AgentService) getPrivilegedTiDBServiceAddress(ctx context.Context, ns string) (string, error) {
	var svcList corev1.ServiceList
	if err := s.dataplaneCli.List(ctx, &svcList,
		client.InNamespace(ns),
		client.MatchingLabels{PingcapComponentLabelKey: PrivilegedTiDBComponentType}); err != nil {
		return "", fmt.Errorf("failed to list services in namespace %s: %w", ns, err)
	}

	if len(svcList.Items) == 0 {
		return "", fmt.Errorf("no privileged TiDB service found in namespace %s", ns)
	}
	if len(svcList.Items) > 1 {
		return "", fmt.Errorf("multiple privileged TiDB services found in namespace %s", ns)
	}

	// as we call the privileged TiDB service from within the cluster agent in the same k8s namespace,
	// we can use the service name directly.
	return svcList.Items[0].Name, nil
}

// errNoRunningPods is returned by findSingleRunningPod when no pod in the given
// list is in Running phase without a DeletionTimestamp. Callers can use errors.Is
// to distinguish this expected "empty" state from genuine errors.
var errNoRunningPods = errors.New("no running pods")

// findSingleRunningPod filters pods to Running phase with no DeletionTimestamp
// (i.e. not terminating) and returns exactly one. During deactivation a
// terminating pod and a new pod can coexist briefly; this filtering avoids
// false "multiple pods" errors.
//
// Returns errNoRunningPods (wrapped) when zero pods match, for callers that
// need to distinguish this from genuine errors via errors.Is.
func findSingleRunningPod(pods []corev1.Pod, ns, component string) (*corev1.Pod, error) {
	var active []*corev1.Pod
	for i := range pods {
		if pods[i].Status.Phase == corev1.PodRunning && pods[i].DeletionTimestamp == nil {
			active = append(active, &pods[i])
		}
	}
	if len(active) == 0 {
		return nil, fmt.Errorf("%w: %s in namespace %s (total: %d)", errNoRunningPods, component, ns, len(pods))
	}
	if len(active) > 1 {
		return nil, fmt.Errorf("multiple running %s pods found in namespace %s", component, ns)
	}
	return active[0], nil
}

func replaceAuditLogStopWait(config string) string {
	// replace any `gracefulStopWaitSeconds: *` to `gracefulStopWaitSeconds: 20`
	re := regexp.MustCompile(`gracefulStopWaitSeconds:\s*\d+`)
	return re.ReplaceAllString(config, "gracefulStopWaitSeconds: 20")
}