package service

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

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

func TestReplaceAuditLogStopWait(t *testing.T) {
	tests := []struct {
		name     string
		config   string
		expected string
	}{
		{
			name:     "simple case",
			config:   "gracefulStopWaitSeconds: 30",
			expected: "gracefulStopWaitSeconds: 20",
		},
		{
			name:     "with extra spaces",
			config:   "gracefulStopWaitSeconds:   60",
			expected: "gracefulStopWaitSeconds: 20",
		},
		{
			name: "multiline config - middle line",
			config: `someOtherConfig: value
gracefulStopWaitSeconds: 45
anotherConfig: value`,
			expected: `someOtherConfig: value
gracefulStopWaitSeconds: 20
anotherConfig: value`,
		},
		{
			name: "multiline config - last line",
			config: `someOtherConfig: value
anotherConfig: value
gracefulStopWaitSeconds: 100`,
			expected: `someOtherConfig: value
anotherConfig: value
gracefulStopWaitSeconds: 20`,
		},
		{
			name: "multiple occurrences",
			config: `gracefulStopWaitSeconds: 30
someConfig: value
gracefulStopWaitSeconds: 60`,
			expected: `gracefulStopWaitSeconds: 20
someConfig: value
gracefulStopWaitSeconds: 20`,
		},
		{
			name:     "no gracefulStopWaitSeconds",
			config:   "someOtherConfig: value",
			expected: "someOtherConfig: value",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := replaceAuditLogStopWait(tt.config)
			if result != tt.expected {
				t.Errorf("replaceAuditLogStopWait() = %q, want %q", result, tt.expected)
			}
		})
	}
}

func TestBuildPrivilegedTiDBPoolURL(t *testing.T) {
	tests := []struct {
		name    string
		host    string
		path    string
		wantURL string
		wantErr bool
	}{
		{
			name:    "valid service host",
			host:    "privileged-tidb",
			path:    "/tidb-pool/status",
			wantURL: "https://privileged-tidb:10080/tidb-pool/status",
			wantErr: false,
		},
		{
			name:    "valid internal dns host",
			host:    "db-0.db-cluster.tidb123.svc",
			path:    "/tidb-pool/activate",
			wantURL: "https://db-0.db-cluster.tidb123.svc:10080/tidb-pool/activate",
			wantErr: false,
		},
		{
			name:    "invalid host with scheme",
			host:    "https://evil.example.com",
			path:    "/tidb-pool/status",
			wantURL: "",
			wantErr: true,
		},
		{
			name:    "invalid host with port",
			host:    "privileged-tidb:10080",
			path:    "/tidb-pool/status",
			wantURL: "",
			wantErr: true,
		},
		{
			name:    "invalid path without slash",
			host:    "privileged-tidb",
			path:    "tidb-pool/status",
			wantURL: "",
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			gotURL, err := buildPrivilegedTiDBPoolURL(tt.host, tt.path)
			if tt.wantErr {
				if err == nil {
					t.Fatalf("expected error, got nil")
				}
				return
			}
			if err != nil {
				t.Fatalf("expected no error, got %v", err)
			}
			if gotURL != tt.wantURL {
				t.Fatalf("unexpected URL, got %q, want %q", gotURL, tt.wantURL)
			}
		})
	}
}

func TestTargetToUsername(t *testing.T) {
	tests := []struct {
		name         string
		target       apiv1.PasswordResetTarget
		wantUsername string
		wantErr      bool
	}{
		{
			name:         "ROOT target",
			target:       apiv1.PasswordResetTarget_ROOT,
			wantUsername: "root",
		},
		{
			name:         "UNSPECIFIED defaults to root",
			target:       apiv1.PasswordResetTarget_PASSWORD_RESET_TARGET_UNSPECIFIED,
			wantUsername: "root",
		},
		{
			name:         "CLOUD_ADMIN target",
			target:       apiv1.PasswordResetTarget_CLOUD_ADMIN,
			wantUsername: "cloud_admin",
		},
		{
			name:         "PLATFORM_ADMIN target",
			target:       apiv1.PasswordResetTarget_PLATFORM_ADMIN,
			wantUsername: "platform_admin",
		},
		{
			name:         "MAINTENANCE_ADMIN target",
			target:       apiv1.PasswordResetTarget_MAINTENANCE_ADMIN,
			wantUsername: "maintenance_admin",
		},
		{
			name:    "invalid target",
			target:  apiv1.PasswordResetTarget(99),
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			username, err := targetToUsername(tt.target)
			if tt.wantErr {
				require.Error(t, err)
				assert.Equal(t, codes.InvalidArgument, status.Code(err))
				return
			}
			require.NoError(t, err)
			assert.Equal(t, tt.wantUsername, username)
		})
	}
}

func TestLogResetPasswordViaPrivilegedTiDBError(t *testing.T) {
	logger := &stubSQLErrorLogger{}
	err := stubSafeLoggableSQLError{
		message: "exec ResetPassword failed",
		cause:   "driver returned ALTER USER 'root'@'%' IDENTIFIED BY 'secret''suffix'",
	}

	msg := logResetPasswordViaPrivilegedTiDBError(logger, err)

	assert.Equal(t, "failed to reset password via privileged TiDB: driver returned ALTER USER 'root'@'%' IDENTIFIED BY '****'", msg)
	assert.Equal(t, []string{
		"failed to reset password via privileged TiDB: driver returned ALTER USER 'root'@'%' IDENTIFIED BY '****'",
	}, logger.messages)
}

func TestAgentService_ResetPassword(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	tests := []struct {
		name          string
		handler       http.HandlerFunc
		deps          *v2HTTPDeps
		depsErr       error
		workerHandler http.HandlerFunc
		workerDeps    *v2HTTPDeps
		workerDepsErr error
		req           *apiv1.ResetPasswordReq
		wantErr       bool
		wantMsg       string
	}{
		{
			name: "V2 success - root",
			handler: func(w http.ResponseWriter, r *http.Request) {
				assert.Equal(t, http.MethodPost, r.Method)
				assert.Equal(t, "/internal/v1/users/root/reset-password", r.URL.Path)
				assert.Equal(t, "application/json", r.Header.Get("Content-Type"))

				body, err := io.ReadAll(r.Body)
				assert.NoError(t, err)
				var payload map[string]string
				assert.NoError(t, json.Unmarshal(body, &payload))
				assert.NotEmpty(t, payload["new_password"])

				w.WriteHeader(http.StatusOK)
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
			},
			wantMsg: "success",
		},
		{
			name: "V2 success - cloud_admin",
			handler: func(w http.ResponseWriter, r *http.Request) {
				assert.Equal(t, "/internal/v1/users/cloud_admin/reset-password", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_CLOUD_ADMIN,
			},
			wantMsg: "success",
		},
		{
			name: "V2 success - platform_admin",
			handler: func(w http.ResponseWriter, r *http.Request) {
				assert.Equal(t, "/internal/v1/users/platform_admin/reset-password", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_PLATFORM_ADMIN,
			},
			wantMsg: "success",
		},
		{
			name: "V2 success - maintenance_admin",
			handler: func(w http.ResponseWriter, r *http.Request) {
				assert.Equal(t, "/internal/v1/users/maintenance_admin/reset-password", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_MAINTENANCE_ADMIN,
			},
			wantMsg: "success",
		},
		{
			name: "V2 success - UNSPECIFIED defaults to root",
			handler: func(w http.ResponseWriter, r *http.Request) {
				assert.Equal(t, "/internal/v1/users/root/reset-password", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_PASSWORD_RESET_TARGET_UNSPECIFIED,
			},
			wantMsg: "success",
		},
		{
			name: "V2 returns 500 - no fallback, returns error",
			handler: func(w http.ResponseWriter, r *http.Request) {
				w.WriteHeader(http.StatusInternalServerError)
				_, _ = w.Write([]byte("internal error"))
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
			},
			wantErr: true,
		},
		{
			name: "V2 returns 502 - no fallback, returns error",
			handler: func(w http.ResponseWriter, r *http.Request) {
				w.WriteHeader(http.StatusBadGateway)
				_, _ = w.Write([]byte("bad gateway"))
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
			},
			wantErr: true,
		},
		{
			name: "password decryption failure",
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: []byte("invalid-encrypted-data"),
				Target:            apiv1.PasswordResetTarget_ROOT,
			},
			wantErr: true,
		},
		{
			name: "empty encrypted password",
			req: &apiv1.ResetPasswordReq{
				ClusterId: "cluster-1",
				Target:    apiv1.PasswordResetTarget_ROOT,
			},
			wantErr: true,
		},
		{
			name: "invalid V2 base URL",
			deps: &v2HTTPDeps{
				client:  &http.Client{},
				baseURL: "http://db-group-tidb.ns.svc:10080",
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
			},
			wantErr: true,
		},
		{
			name:    "address resolution failure",
			depsErr: fmt.Errorf("failed to resolve TiDB address: no tidb group found"),
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
			},
			wantErr: true,
		},
		{
			name: "invalid target returns error",
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget(99),
			},
			wantErr: true,
		},
		{
			name: "V2 success - worker TiDB root",
			workerHandler: func(w http.ResponseWriter, r *http.Request) {
				assert.Equal(t, http.MethodPost, r.Method)
				assert.Equal(t, "/internal/v1/users/root/reset-password", r.URL.Path)
				w.WriteHeader(http.StatusOK)
			},
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
				UseTidbWorker:     true,
			},
			wantMsg: "success",
		},
		{
			name:          "worker TiDB address resolution failure",
			workerDepsErr: fmt.Errorf("failed to resolve worker TiDB address: no worker TiDB group found"),
			req: &apiv1.ResetPasswordReq{
				ClusterId:         "cluster-1",
				EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
				Target:            apiv1.PasswordResetTarget_ROOT,
				UseTidbWorker:     true,
			},
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			depsResolver := func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
				return nil, fmt.Errorf("resolver should not be called")
			}
			switch {
			case tt.deps != nil:
				depsResolver = func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
					return tt.deps, nil
				}
			case tt.depsErr != nil:
				depsResolver = func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
					return nil, tt.depsErr
				}
			case tt.handler != nil:
				ts := httptest.NewTLSServer(tt.handler)
				t.Cleanup(ts.Close)

				depsResolver = func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
					return &v2HTTPDeps{
						client:  ts.Client(),
						baseURL: ts.URL,
					}, nil
				}
			}

			workerDepsResolver := func(_ context.Context, _ client.Client) (*v2HTTPDeps, error) {
				return nil, fmt.Errorf("worker resolver should not be called")
			}
			switch {
			case tt.workerDeps != nil:
				workerDepsResolver = func(_ context.Context, _ client.Client) (*v2HTTPDeps, error) {
					return tt.workerDeps, nil
				}
			case tt.workerDepsErr != nil:
				workerDepsResolver = func(_ context.Context, _ client.Client) (*v2HTTPDeps, error) {
					return nil, tt.workerDepsErr
				}
			case tt.workerHandler != nil:
				wts := httptest.NewTLSServer(tt.workerHandler)
				t.Cleanup(wts.Close)

				workerDepsResolver = func(_ context.Context, _ client.Client) (*v2HTTPDeps, error) {
					return &v2HTTPDeps{
						client:  wts.Client(),
						baseURL: wts.URL,
					}, nil
				}
			}

			svc := &AgentService{
				encryptor:             encryptor,
				resolveV2HTTPDeps:     depsResolver,
				resolveWorkerHTTPDeps: workerDepsResolver,
			}

			resp, err := svc.ResetPassword(context.Background(), tt.req)
			if tt.wantErr {
				require.Error(t, err)
				return
			}

			require.NoError(t, err)
			require.NotNil(t, resp)
			assert.Equal(t, tt.wantMsg, resp.Message)
		})
	}
}

func TestAgentService_ResetRootPassword_DelegatesToResetPassword(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	handler := func(w http.ResponseWriter, r *http.Request) {
		assert.Equal(t, http.MethodPost, r.Method)
		assert.Equal(t, "/internal/v1/users/root/reset-password", r.URL.Path)
		w.WriteHeader(http.StatusOK)
	}

	ts := httptest.NewTLSServer(http.HandlerFunc(handler))
	t.Cleanup(ts.Close)

	svc := &AgentService{
		encryptor: encryptor,
		resolveV2HTTPDeps: func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
			return &v2HTTPDeps{
				client:  ts.Client(),
				baseURL: ts.URL,
			}, nil
		},
	}

	resp, err := svc.ResetRootPassword(context.Background(), &apiv1.ResetRootPasswordReq{
		ClusterId:         "cluster-1",
		EncryptedPassword: encryptForTest(t, encryptor, "password123"),
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "success", resp.Message)
}

func TestAgentService_ResetPassword_404FallbackWorkerUsesSYSTEMKeyspace(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	// Worker TiDB HTTP API returns 404 -> fallback to V1 privileged TiDB
	workerTS := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		assert.Equal(t, http.MethodPost, r.Method)
		assert.Equal(t, "/internal/v1/users/root/reset-password", r.URL.Path)
		w.WriteHeader(http.StatusNotFound)
	}))
	t.Cleanup(workerTS.Close)

	stubExec := &stubPrivilegedExecutor{}
	var capturedLifecycle *privilegedTiDBLifecycle

	oldFactory := newPrivilegedTiDBExecutor
	newPrivilegedTiDBExecutor = func(lifecycle sqlexec.PrivilegedLifecycle) sqlexec.Executor {
		lc, ok := lifecycle.(*privilegedTiDBLifecycle)
		require.True(t, ok)
		capturedLifecycle = lc
		return stubExec
	}
	t.Cleanup(func() {
		newPrivilegedTiDBExecutor = oldFactory
	})

	svc := &AgentService{
		encryptor: encryptor,
		resolveWorkerHTTPDeps: func(_ context.Context, _ client.Client) (*v2HTTPDeps, error) {
			return &v2HTTPDeps{
				client:  workerTS.Client(),
				baseURL: workerTS.URL,
			}, nil
		},
	}

	resp, err := svc.ResetPassword(context.Background(), &apiv1.ResetPasswordReq{
		ClusterId:         "cluster-1",
		EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
		Target:            apiv1.PasswordResetTarget_ROOT,
		UseTidbWorker:     true,
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "success", resp.Message)

	// Verify V1 fallback used "SYSTEM" keyspace (not "cluster-1")
	require.Len(t, stubExec.calls, 1)
	assert.Equal(t, "SYSTEM", stubExec.calls[0].clusterID)
	assert.Equal(t, "ResetPassword", stubExec.calls[0].cmdName)

	require.NotNil(t, capturedLifecycle)
	assert.Equal(t, "maintenance_admin", capturedLifecycle.username)
	assert.True(t, capturedLifecycle.enableAuditlog)
}

// TestAgentService_ResetPassword_V2_404_V1Occupied_V0Success exercises the
// full fallback chain: TiDB HTTP API returns 404, the privileged TiDB is
// activated for a different keyspace (V1 fails fast with "not available"),
// and the V0 direct-MySQL fallback is invoked and succeeds. This is the
// dominant production path for "concurrent cluster create with old TiDB
// images" -- removing the V1 retry that used to drain ctx is what lets V0
// reach this success terminus.
func TestAgentService_ResetPassword_V2_404_V1Occupied_V0Success(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	// V2: returns 404 -> fallback to V1.
	v2TS := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		assert.Equal(t, http.MethodPost, r.Method)
		assert.Equal(t, "/internal/v1/users/cloud_admin/reset-password", r.URL.Path)
		w.WriteHeader(http.StatusNotFound)
	}))
	t.Cleanup(v2TS.Close)

	// V1: privileged TiDB /tidb-pool/status reports "activated for another
	// keyspace" -> isPrivilegedTiDBAvailable returns false, prepare fails fast.
	const otherKeyspace = "other-cluster"
	var v1StatusCalls atomic.Int32
	v1TS := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		v1StatusCalls.Add(1)
		assert.Equal(t, "/tidb-pool/status", r.URL.Path,
			"only the status endpoint should be hit; activate must NOT fire when occupied")
		_ = json.NewEncoder(w).Encode(StandbyTiDBStatus{State: "activated", KeyspaceName: otherKeyspace})
	}))
	t.Cleanup(v1TS.Close)
	overrideBuildURL(t, v1TS.URL)

	// Fake k8s: TLS secret + privileged TiDB Service + a Running privileged TiDB
	// pod, all of which prepare() needs before it can fetch /tidb-pool/status.
	const namespace = "tidb-pool"
	t.Setenv("NAMESPACE", namespace)

	certPEM, keyPEM := generateTLSPEMForTest(t)
	tlsSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{Name: tlsClientSecret, Namespace: namespace},
		Data: map[string][]byte{
			corev1.ServiceAccountRootCAKey: certPEM,
			corev1.TLSCertKey:              certPEM,
			corev1.TLSPrivateKeyKey:        keyPEM,
		},
	}
	privSvc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "privileged-tidb",
			Namespace: namespace,
			Labels:    map[string]string{PingcapComponentLabelKey: PrivilegedTiDBComponentType},
		},
	}
	privPod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)
	// Pod for the "other" cluster so the preempt step considers the activation
	// legitimate (a live cluster is using it) and skips the stale-state fast-path.
	// Without this, preempt would force-delete the privileged TiDB pod and then
	// poll for ~15s waiting for it to come back, slowing the test pointlessly.
	otherPod := newClusterTiDBPod("other-tidb-0", namespace, otherKeyspace, corev1.PodRunning)
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tlsSecret, privSvc, privPod, otherPod).Build()

	// V0: succeeds.
	var v0Called bool
	oldDirect := resetPasswordDirectFunc
	resetPasswordDirectFunc = func(_ *AgentService, _ context.Context, clusterID, username, _ string, _ bool) error {
		v0Called = true
		assert.Equal(t, "my-cluster", clusterID)
		assert.Equal(t, "cloud_admin", username)
		return nil
	}
	t.Cleanup(func() { resetPasswordDirectFunc = oldDirect })

	svc := &AgentService{
		dataplaneCli: cli,
		encryptor:    encryptor,
		resolveV2HTTPDeps: func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
			return &v2HTTPDeps{client: v2TS.Client(), baseURL: v2TS.URL}, nil
		},
	}

	resp, err := svc.ResetPassword(context.Background(), &apiv1.ResetPasswordReq{
		ClusterId:         "my-cluster",
		EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
		Target:            apiv1.PasswordResetTarget_CLOUD_ADMIN,
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "success", resp.Message)

	assert.GreaterOrEqual(t, int(v1StatusCalls.Load()), 1, "V1 must consult /tidb-pool/status at least once")
	assert.True(t, v0Called, "V0 fallback must run after V1 reports occupied")
}

func TestAgentService_ResetPassword_V0FallbackOnV1Failure(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	// V2: returns 404 to trigger fallback
	ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusNotFound)
	}))
	t.Cleanup(ts.Close)

	// V1: fails
	stubExec := &stubPrivilegedExecutor{err: fmt.Errorf("privileged TiDB unavailable")}
	oldFactory := newPrivilegedTiDBExecutor
	newPrivilegedTiDBExecutor = func(lifecycle sqlexec.PrivilegedLifecycle) sqlexec.Executor {
		return stubExec
	}
	t.Cleanup(func() { newPrivilegedTiDBExecutor = oldFactory })

	// V0: succeeds, capture params
	var v0ClusterID, v0Username string
	var v0UseTiDBWorker bool
	var v0Called bool
	oldDirect := resetPasswordDirectFunc
	resetPasswordDirectFunc = func(_ *AgentService, _ context.Context, clusterID, username, _ string, useTiDBWorker bool) error {
		v0Called = true
		v0ClusterID = clusterID
		v0Username = username
		v0UseTiDBWorker = useTiDBWorker
		return nil
	}
	t.Cleanup(func() { resetPasswordDirectFunc = oldDirect })

	svc := &AgentService{
		encryptor: encryptor,
		resolveV2HTTPDeps: func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
			return &v2HTTPDeps{
				client:  ts.Client(),
				baseURL: ts.URL,
			}, nil
		},
	}

	resp, err := svc.ResetPassword(context.Background(), &apiv1.ResetPasswordReq{
		ClusterId:         "cluster-1",
		EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
		Target:            apiv1.PasswordResetTarget_ROOT,
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "success", resp.Message)
	assert.True(t, v0Called)
	assert.Equal(t, "cluster-1", v0ClusterID)
	assert.Equal(t, "root", v0Username)
	assert.False(t, v0UseTiDBWorker)
}

func TestAgentService_ResetPassword_AllMethodsFail(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	// V2: returns 404 to trigger fallback
	ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusNotFound)
	}))
	t.Cleanup(ts.Close)

	// V1: fails
	stubExec := &stubPrivilegedExecutor{err: fmt.Errorf("privileged TiDB unavailable")}
	oldFactory := newPrivilegedTiDBExecutor
	newPrivilegedTiDBExecutor = func(lifecycle sqlexec.PrivilegedLifecycle) sqlexec.Executor {
		return stubExec
	}
	t.Cleanup(func() { newPrivilegedTiDBExecutor = oldFactory })

	// V0: also fails
	oldDirect := resetPasswordDirectFunc
	resetPasswordDirectFunc = func(_ *AgentService, _ context.Context, _, _, _ string, _ bool) error {
		return fmt.Errorf("direct MySQL connection refused")
	}
	t.Cleanup(func() { resetPasswordDirectFunc = oldDirect })

	svc := &AgentService{
		encryptor: encryptor,
		resolveV2HTTPDeps: func(_ context.Context, _ client.Client, _ string) (*v2HTTPDeps, error) {
			return &v2HTTPDeps{
				client:  ts.Client(),
				baseURL: ts.URL,
			}, nil
		},
	}

	resp, err := svc.ResetPassword(context.Background(), &apiv1.ResetPasswordReq{
		ClusterId:         "cluster-1",
		EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
		Target:            apiv1.PasswordResetTarget_ROOT,
	})
	require.Error(t, err)
	require.NotNil(t, resp)
	assert.Contains(t, resp.Message, "all password reset methods failed")
	assert.Contains(t, resp.Message, "V0 direct MySQL")
}

func TestAgentService_ResetPassword_V0FallbackWorkerTiDB(t *testing.T) {
	encryptor, err := agentcrypto.NewEncryptor()
	require.NoError(t, err)

	// V2: returns 404 to trigger fallback (worker TiDB)
	workerTS := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusNotFound)
	}))
	t.Cleanup(workerTS.Close)

	// V1: fails
	stubExec := &stubPrivilegedExecutor{err: fmt.Errorf("privileged TiDB unavailable")}
	oldFactory := newPrivilegedTiDBExecutor
	newPrivilegedTiDBExecutor = func(lifecycle sqlexec.PrivilegedLifecycle) sqlexec.Executor {
		return stubExec
	}
	t.Cleanup(func() { newPrivilegedTiDBExecutor = oldFactory })

	// V0: succeeds, capture useTiDBWorker
	var v0UseTiDBWorker bool
	var v0Called bool
	oldDirect := resetPasswordDirectFunc
	resetPasswordDirectFunc = func(_ *AgentService, _ context.Context, _, _, _ string, useTiDBWorker bool) error {
		v0Called = true
		v0UseTiDBWorker = useTiDBWorker
		return nil
	}
	t.Cleanup(func() { resetPasswordDirectFunc = oldDirect })

	svc := &AgentService{
		encryptor: encryptor,
		resolveWorkerHTTPDeps: func(_ context.Context, _ client.Client) (*v2HTTPDeps, error) {
			return &v2HTTPDeps{
				client:  workerTS.Client(),
				baseURL: workerTS.URL,
			}, nil
		},
	}

	resp, err := svc.ResetPassword(context.Background(), &apiv1.ResetPasswordReq{
		ClusterId:         "cluster-1",
		EncryptedPassword: encryptForTest(t, encryptor, "new_password_123"),
		Target:            apiv1.PasswordResetTarget_ROOT,
		UseTidbWorker:     true,
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, "success", resp.Message)
	assert.True(t, v0Called)
	assert.True(t, v0UseTiDBWorker)
}

// overrideBuildURL redirects buildPrivilegedTiDBPoolURL to use a test server URL.
func overrideBuildURL(t *testing.T, serverURL string) {
	t.Helper()
	orig := buildPrivilegedTiDBPoolURL
	buildPrivilegedTiDBPoolURL = func(_, path string) (string, error) {
		return serverURL + path, nil
	}
	t.Cleanup(func() { buildPrivilegedTiDBPoolURL = orig })
}

func TestIsPrivilegedTiDBAvailable_Standby(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = json.NewEncoder(w).Encode(StandbyTiDBStatus{State: "standby"})
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)

	svc := &AgentService{}
	available, err := svc.isPrivilegedTiDBAvailable(context.Background(), http.Client{}, "test-host", "cluster-1")
	require.NoError(t, err)
	assert.True(t, available)
}

func TestIsPrivilegedTiDBAvailable_OccupiedByAnotherKeyspace(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = json.NewEncoder(w).Encode(StandbyTiDBStatus{State: "activated", KeyspaceName: "other-cluster"})
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)

	svc := &AgentService{}
	available, err := svc.isPrivilegedTiDBAvailable(context.Background(), http.Client{}, "test-host", "cluster-1")
	require.NoError(t, err)
	assert.False(t, available, "must report unavailable when activated for a different keyspace")
}

func TestIsPrivilegedTiDBAvailable_ActivatedForSameKeyspace(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = json.NewEncoder(w).Encode(StandbyTiDBStatus{State: "activated", KeyspaceName: "cluster-1"})
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)

	svc := &AgentService{}
	available, err := svc.isPrivilegedTiDBAvailable(context.Background(), http.Client{}, "test-host", "cluster-1")
	require.NoError(t, err)
	assert.True(t, available)
}

// newPrivilegedTiDBPod returns a privileged-tidb labelled pod in the given phase.
func newPrivilegedTiDBPod(name string, uid types.UID, phase corev1.PodPhase, terminating bool) *corev1.Pod {
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "tidb-pool",
			UID:       uid,
			Labels:    map[string]string{PingcapComponentLabelKey: PrivilegedTiDBComponentType},
		},
		Status: corev1.PodStatus{Phase: phase},
	}
	if terminating {
		now := metav1.NewTime(time.Now())
		pod.DeletionTimestamp = &now
		pod.Finalizers = []string{"kubernetes"}
	}
	return pod
}

func newFakeClientWithPods(t *testing.T, pods ...*corev1.Pod) client.Client {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	builder := fake.NewClientBuilder().WithScheme(scheme)
	for _, p := range pods {
		builder = builder.WithObjects(p)
	}
	return builder.Build()
}

func TestFindSingleRunningPod_NoRunningPods_WrapsSentinel(t *testing.T) {
	// Only terminating pod present -- filtered out.
	pods := []corev1.Pod{*newPrivilegedTiDBPod("p1", "u1", corev1.PodRunning, true)}
	_, err := findSingleRunningPod(pods, "tidb-pool", PrivilegedTiDBComponentType)
	require.Error(t, err)
	assert.ErrorIs(t, err, errNoRunningPods)
}

func TestFindSingleRunningPod_Empty_WrapsSentinel(t *testing.T) {
	_, err := findSingleRunningPod(nil, "tidb-pool", PrivilegedTiDBComponentType)
	require.Error(t, err)
	assert.ErrorIs(t, err, errNoRunningPods)
}

func TestFindSingleRunningPod_Multiple_NotSentinel(t *testing.T) {
	pods := []corev1.Pod{
		*newPrivilegedTiDBPod("p1", "u1", corev1.PodRunning, false),
		*newPrivilegedTiDBPod("p2", "u2", corev1.PodRunning, false),
	}
	_, err := findSingleRunningPod(pods, "tidb-pool", PrivilegedTiDBComponentType)
	require.Error(t, err)
	require.NotErrorIs(t, err, errNoRunningPods)
	assert.Contains(t, err.Error(), "multiple")
}

func TestDeactivatePrivilegedTiDB_UIDMatches_DeletesPod(t *testing.T) {
	pod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)
	cli := newFakeClientWithPods(t, pod)
	svc := &AgentService{dataplaneCli: cli}

	require.NoError(t, svc.deactivatePrivilegedTiDB("tidb-pool", "uid-A"))

	var got corev1.Pod
	err := cli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-0"}, &got)
	assert.True(t, apierrors.IsNotFound(err), "pod should be deleted when UID matches")
}

func TestDeactivatePrivilegedTiDB_UIDMismatch_SkipsDelete(t *testing.T) {
	// The running pod has a different UID -- it's a replacement pod activated by someone else.
	// Our deactivation must NOT delete it.
	pod := newPrivilegedTiDBPod("priv-1", "uid-B", corev1.PodRunning, false)
	cli := newFakeClientWithPods(t, pod)
	svc := &AgentService{dataplaneCli: cli}

	require.NoError(t, svc.deactivatePrivilegedTiDB("tidb-pool", "uid-A"))

	var got corev1.Pod
	err := cli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-1"}, &got)
	assert.NoError(t, err, "replacement pod must be left alone")
}

func TestDeactivatePrivilegedTiDB_NoRunningPods_ReturnsNil(t *testing.T) {
	// Only a terminating pod exists -- findSingleRunningPod returns errNoRunningPods.
	// deactivate should treat this as success (the controller will recreate a standby pod).
	pod := newPrivilegedTiDBPod("priv-old", "uid-X", corev1.PodRunning, true)
	cli := newFakeClientWithPods(t, pod)
	svc := &AgentService{dataplaneCli: cli}

	assert.NoError(t, svc.deactivatePrivilegedTiDB("tidb-pool", "uid-X"))
}

func TestDeactivatePrivilegedTiDB_EmptyUID_FallsBackToLegacy(t *testing.T) {
	// Empty expectedUID preserves the legacy behaviour of deleting whichever
	// running pod is found (for call sites that couldn't capture the UID).
	pod := newPrivilegedTiDBPod("priv-any", "uid-Z", corev1.PodRunning, false)
	cli := newFakeClientWithPods(t, pod)
	svc := &AgentService{dataplaneCli: cli}

	require.NoError(t, svc.deactivatePrivilegedTiDB("tidb-pool", ""))

	var got corev1.Pod
	err := cli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-any"}, &got)
	assert.True(t, apierrors.IsNotFound(err), "legacy fallback should delete the running pod")
}

func TestGetPrivilegedTiDBPodUID_ReturnsUID(t *testing.T) {
	pod := newPrivilegedTiDBPod("priv-live", "uid-Q", corev1.PodRunning, false)
	cli := newFakeClientWithPods(t, pod)
	svc := &AgentService{dataplaneCli: cli}

	uid, err := svc.getPrivilegedTiDBPodUID(context.Background(), "tidb-pool")
	require.NoError(t, err)
	assert.Equal(t, types.UID("uid-Q"), uid)
}

func TestGetPrivilegedTiDBPodUID_NoRunningPod_WrapsSentinel(t *testing.T) {
	cli := newFakeClientWithPods(t)
	svc := &AgentService{dataplaneCli: cli}

	_, err := svc.getPrivilegedTiDBPodUID(context.Background(), "tidb-pool")
	require.Error(t, err)
	assert.ErrorIs(t, err, errNoRunningPods)
}