package service

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

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	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"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

// newClusterTiDBPod returns a Pod that mimics a real cluster's TiDB pod so the
// fake K8s client can find it via List with label selectors.
func newClusterTiDBPod(name, namespace, clusterID string, phase corev1.PodPhase) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
			Labels: map[string]string{
				"pingcap.com/component":      "tidb",
				"tags.tidbcloud.com/cluster": clusterID,
			},
		},
		Status: corev1.PodStatus{Phase: phase},
	}
}

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

	stale, err := svc.isActivatedKeyspaceStale(context.Background(), "tidb-pool", "deleted-cluster")
	require.NoError(t, err)
	assert.True(t, stale, "no matching pods means cluster is gone")
}

func TestIsActivatedKeyspaceStale_ReturnsFalseWhenPodsExist(t *testing.T) {
	pod := newClusterTiDBPod("tidb-0", "tidb-pool", "live-cluster", corev1.PodRunning)
	cli := newFakeClientWithPods(t, pod)
	svc := &AgentService{dataplaneCli: cli}

	stale, err := svc.isActivatedKeyspaceStale(context.Background(), "tidb-pool", "live-cluster")
	require.NoError(t, err)
	assert.False(t, stale, "live cluster's pods must keep activation alive")
}

func TestIsActivatedKeyspaceStale_AnyPodPhaseCounts(t *testing.T) {
	// Pending or Terminating still counts as "cluster exists" -- avoids
	// preempting during transient zero-Running windows.
	pending := newClusterTiDBPod("tidb-0", "tidb-pool", "starting-cluster", corev1.PodPending)
	cli := newFakeClientWithPods(t, pending)
	svc := &AgentService{dataplaneCli: cli}

	stale, err := svc.isActivatedKeyspaceStale(context.Background(), "tidb-pool", "starting-cluster")
	require.NoError(t, err)
	assert.False(t, stale)
}

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

	stale, err := svc.isActivatedKeyspaceStale(context.Background(), "tidb-pool", systemKeyspace)
	require.NoError(t, err)
	assert.False(t, stale, "system keyspace must never be reported stale")
}

func TestIsActivatedKeyspaceStale_OtherClusterPodsDoNotCount(t *testing.T) {
	other := newClusterTiDBPod("tidb-0", "tidb-pool", "other-cluster", corev1.PodRunning)
	cli := newFakeClientWithPods(t, other)
	svc := &AgentService{dataplaneCli: cli}

	stale, err := svc.isActivatedKeyspaceStale(context.Background(), "tidb-pool", "missing-cluster")
	require.NoError(t, err)
	assert.True(t, stale, "label selector must filter by cluster id")
}

// preemptHarness wires an httptest.Server returning a fixed status into the
// fetchPrivilegedTiDBStatus path, plus a fake K8s client seeded with pods.
func preemptHarness(t *testing.T, status StandbyTiDBStatus, pods ...*corev1.Pod) *AgentService {
	t.Helper()
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = json.NewEncoder(w).Encode(status)
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)
	return &AgentService{dataplaneCli: newFakeClientWithPods(t, pods...)}
}

func TestPreemptStalePrivilegedTiDB_StandbyNoOp(t *testing.T) {
	svc := preemptHarness(t, StandbyTiDBStatus{State: "standby"})

	preempted, err := svc.preemptStalePrivilegedTiDB(
		context.Background(), http.Client{}, "test-host", "tidb-pool", "my-cluster")
	require.NoError(t, err)
	assert.False(t, preempted)
}

func TestPreemptStalePrivilegedTiDB_OurClusterNoOp(t *testing.T) {
	svc := preemptHarness(t, StandbyTiDBStatus{State: "activated", KeyspaceName: "my-cluster"})

	preempted, err := svc.preemptStalePrivilegedTiDB(
		context.Background(), http.Client{}, "test-host", "tidb-pool", "my-cluster")
	require.NoError(t, err)
	assert.False(t, preempted, "must not preempt activations bound to our own cluster")
}

func TestPreemptStalePrivilegedTiDB_SystemKeyspaceNoOp(t *testing.T) {
	// SYSTEM activation must never be preempted at runtime: the deactivation
	// goroutine runs after the mutex is released (30s grace window), so holding
	// the mutex does not imply the goroutine was lost. Startup recovery handles
	// the restart scenario instead.
	svc := preemptHarness(t, StandbyTiDBStatus{State: "activated", KeyspaceName: systemKeyspace})

	preempted, err := svc.preemptStalePrivilegedTiDB(
		context.Background(), http.Client{}, "test-host", "tidb-pool", "my-cluster")
	require.NoError(t, err)
	assert.False(t, preempted, "must never preempt the system keyspace activation at runtime")
}

func TestPreemptStalePrivilegedTiDB_LiveOtherClusterNoOp(t *testing.T) {
	otherPod := newClusterTiDBPod("tidb-0", "tidb-pool", "other-cluster", corev1.PodRunning)
	svc := preemptHarness(t, StandbyTiDBStatus{State: "activated", KeyspaceName: "other-cluster"}, otherPod)

	// Also seed the privileged TiDB pod so the (would-be) deactivation has a target;
	// we assert it is NOT deleted.
	privPod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)
	require.NoError(t, svc.dataplaneCli.Create(context.Background(), privPod))

	preempted, err := svc.preemptStalePrivilegedTiDB(
		context.Background(), http.Client{}, "test-host", "tidb-pool", "my-cluster")
	require.NoError(t, err)
	assert.False(t, preempted, "must not preempt while the activated cluster is alive")

	var got corev1.Pod
	err = svc.dataplaneCli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-0"}, &got)
	assert.NoError(t, err, "privileged TiDB pod must remain")
}

func TestPreemptStalePrivilegedTiDB_StaleClusterDeletesPod(t *testing.T) {
	// Activated for keyspace=ghost, but no TiDB pods labelled with that cluster.
	svc := preemptHarness(t, StandbyTiDBStatus{State: "activated", KeyspaceName: "ghost"})
	privPod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)
	require.NoError(t, svc.dataplaneCli.Create(context.Background(), privPod))

	preempted, err := svc.preemptStalePrivilegedTiDB(
		context.Background(), http.Client{}, "test-host", "tidb-pool", "my-cluster")
	require.NoError(t, err)
	assert.True(t, preempted, "deleted-cluster activation must be preempted")

	var got corev1.Pod
	err = svc.dataplaneCli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-0"}, &got)
	assert.True(t, apierrors.IsNotFound(err), "privileged TiDB pod must be deleted to allow recreate as standby")
}

func TestPreemptStalePrivilegedTiDB_StatusFetchErrorReturnsError(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusInternalServerError)
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)
	svc := &AgentService{dataplaneCli: newFakeClientWithPods(t)}

	preempted, err := svc.preemptStalePrivilegedTiDB(
		context.Background(), http.Client{}, "test-host", "tidb-pool", "my-cluster")
	require.Error(t, err)
	assert.False(t, preempted)
}

// TestPrepareForPrivilegedTiDBExecution_StaleOwnerPreemptThenWait verifies the
// end-to-end wiring: when the privileged TiDB is wedged on a deleted cluster's
// keyspace, prepareForPrivilegedTiDBExecution must (1) detect the stale state,
// (2) force-delete the privileged TiDB pod, (3) the subsequent availability
// check sees standby (the operator would have recreated the pod), (4) activate
// proceeds. The counter on the status handler doubles as the simulator for
// "pod deleted -> recreated as standby".
func TestPrepareForPrivilegedTiDBExecution_StaleOwnerPreemptThenWait(t *testing.T) {
	const (
		namespace  = "tidb-pool"
		myCluster  = "my-cluster"
		ghostOwner = "ghost-cluster" // deleted; no TiDB pods labelled with it
	)

	// HTTP server: stateful status (first activated->ghost, then standby) +
	// activate endpoint that records the keyspace it was asked to bind.
	var statusCalls, activateCalls atomic.Int32
	var activatedFor atomic.Value
	activatedFor.Store("")

	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/tidb-pool/status":
			n := statusCalls.Add(1)
			if n == 1 {
				_ = json.NewEncoder(w).Encode(StandbyTiDBStatus{State: "activated", KeyspaceName: ghostOwner})
				return
			}
			_ = json.NewEncoder(w).Encode(StandbyTiDBStatus{State: "standby"})
		case "/tidb-pool/activate":
			activateCalls.Add(1)
			var body map[string]string
			_ = json.NewDecoder(r.Body).Decode(&body)
			activatedFor.Store(body["keyspace_name"])
			w.WriteHeader(http.StatusOK)
		default:
			http.NotFound(w, r)
		}
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)

	// K8s fake: TLS secret (for GetTLSConfig) + a Running privileged-tidb pod
	// that the preempt step will delete. NO TiDB pods labelled with ghostOwner
	// (that absence is the stale signal).
	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,
		},
	}
	privPod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tlsSecret, privPod).Build()

	svc := &AgentService{dataplaneCli: cli}

	// enableAuditlog=false skips the getTiDBAuditConfig path (which would need
	// a separate cluster-tagged TiDB pod). The preempt + availability check +
	// activate orchestration is unchanged.
	err := svc.prepareForPrivilegedTiDBExecution(
		context.Background(), myCluster, namespace, "test-host", false)
	require.NoError(t, err)

	// Preempt must have run (status fetched first time): privileged TiDB pod is gone.
	var got corev1.Pod
	getErr := cli.Get(context.Background(), client.ObjectKey{Namespace: namespace, Name: "priv-0"}, &got)
	assert.True(t, apierrors.IsNotFound(getErr),
		"expected privileged TiDB pod to be force-deleted by preempt, got err=%v", getErr)

	// Status must have been fetched twice (preempt + availability check); activate fired exactly once.
	assert.GreaterOrEqual(t, int(statusCalls.Load()), 2,
		"status must be fetched in preempt AND in availability check (got %d)", statusCalls.Load())
	assert.Equal(t, int32(1), activateCalls.Load(), "activate must fire exactly once after preempt")
	assert.Equal(t, myCluster, activatedFor.Load().(string), "activate must bind to our clusterID")
}

// startupRecoveryHarness builds an AgentService wired to a fake K8s client and
// a test HTTP server for startup-recovery tests. NAMESPACE is set to "tidb-pool"
// for the duration of the test.
func startupRecoveryHarness(t *testing.T, statusToReturn StandbyTiDBStatus, privPod *corev1.Pod) *AgentService {
	t.Helper()
	t.Setenv("NAMESPACE", "tidb-pool")

	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = json.NewEncoder(w).Encode(statusToReturn)
	}))
	t.Cleanup(ts.Close)
	overrideBuildURL(t, ts.URL)

	certPEM, keyPEM := generateTLSPEMForTest(t)
	tlsSecret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{Name: tlsClientSecret, Namespace: "tidb-pool"},
		Data: map[string][]byte{
			corev1.ServiceAccountRootCAKey: certPEM,
			corev1.TLSCertKey:              certPEM,
			corev1.TLSPrivateKeyKey:        keyPEM,
		},
	}
	privSvc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "db-test-privileged-tidb",
			Namespace: "tidb-pool",
			Labels:    map[string]string{PingcapComponentLabelKey: PrivilegedTiDBComponentType},
		},
	}

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	builder := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tlsSecret, privSvc)
	if privPod != nil {
		builder = builder.WithObjects(privPod)
	}
	return &AgentService{dataplaneCli: builder.Build()}
}

func TestRecoverPrivilegedTiDBOnStartup_SystemStuck(t *testing.T) {
	privPod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)
	svc := startupRecoveryHarness(t, StandbyTiDBStatus{State: "activated", KeyspaceName: systemKeyspace}, privPod)

	svc.RecoverPrivilegedTiDBOnStartup(context.Background())

	var got corev1.Pod
	err := svc.dataplaneCli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-0"}, &got)
	assert.True(t, apierrors.IsNotFound(err), "privileged TiDB pod must be deleted on startup recovery")
}

func TestRecoverPrivilegedTiDBOnStartup_StandbyNoOp(t *testing.T) {
	privPod := newPrivilegedTiDBPod("priv-0", "uid-A", corev1.PodRunning, false)
	svc := startupRecoveryHarness(t, StandbyTiDBStatus{State: "standby"}, privPod)

	svc.RecoverPrivilegedTiDBOnStartup(context.Background())

	var got corev1.Pod
	err := svc.dataplaneCli.Get(context.Background(), client.ObjectKey{Namespace: "tidb-pool", Name: "priv-0"}, &got)
	assert.NoError(t, err, "privileged TiDB pod must remain when status is standby")
}