package common

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	helmaction "helm.sh/helm/v3/pkg/action"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/rest"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	ticdcpb "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
)

const (
	testCDCClusterName = "123456789"
	testCDCDesiredVer  = "1.27"
	testCDCTiCDCID     = "cdc-a"
	testCDCTiCDCID2    = "cdc-b"
)

// stubEnvProvider implements dataplane.EnvProvider for CDC upgrade tests.
type stubEnvProvider struct {
	dpCli client.Client
}

func (s *stubEnvProvider) GetClient() client.Client                 { return s.dpCli }
func (s *stubEnvProvider) GetHelmConfig() *helmaction.Configuration { return nil }
func (s *stubEnvProvider) GetRestConfig() *rest.Config              { return nil }
func (s *stubEnvProvider) GetRegion() string                        { return "" }
func (s *stubEnvProvider) GetClusterName() string                   { return "" }
func (s *stubEnvProvider) GetClusterEndpoint() string               { return "" }
func (s *stubEnvProvider) GetClusterCA() []byte                     { return nil }
func (s *stubEnvProvider) GetAccount() string                       { return "" }
func (s *stubEnvProvider) GetNetwork() string                       { return "" }
func (s *stubEnvProvider) AWS() dataplane.AWSInterface              { return nil }
func (s *stubEnvProvider) GCP() dataplane.GCPInterface              { return nil }

var _ dataplane.EnvProvider = (*stubEnvProvider)(nil)

type recordingTiCDCClient struct {
	listResp           *ticdcpb.ListTiCDCClustersResponse
	listErr            error
	updateErr          error
	lastTidbClusterID  string
	lastTicdcFilter    []string
	updatedNodeVersion map[string]string
}

func (c *recordingTiCDCClient) ListTiCDCClusters(_ context.Context, tidbClusterID string, ticdcClusters []string) (*ticdcpb.ListTiCDCClustersResponse, error) {
	c.lastTidbClusterID = tidbClusterID
	c.lastTicdcFilter = append([]string(nil), ticdcClusters...)
	if c.listErr != nil {
		return nil, c.listErr
	}
	return c.listResp, nil
}

func (c *recordingTiCDCClient) UpdateTiCDCNodeVersion(_ context.Context, ticdcClusterID, nodeVersion string) error {
	if c.updatedNodeVersion == nil {
		c.updatedNodeVersion = make(map[string]string)
	}
	c.updatedNodeVersion[ticdcClusterID] = nodeVersion
	return c.updateErr
}

func newTestCDCUpgrader(t *testing.T, ticdcCli *recordingTiCDCClient, uo *cloudv1.UpgradeOperation) (*CDCUpgrader, client.Client) {
	t.Helper()
	scheme := runtime.NewScheme()
	if err := cloudv1.AddToScheme(scheme); err != nil {
		t.Fatalf("add cloudv1 scheme: %v", err)
	}
	if err := corev1.AddToScheme(scheme); err != nil {
		t.Fatalf("add corev1 scheme: %v", err)
	}
	infraCli := fake.NewClientBuilder().
		WithScheme(scheme).
		WithStatusSubresource(&cloudv1.UpgradeOperation{}).
		WithObjects(uo).
		Build()
	return NewCDCUpgrader(infraCli, ticdcCli), infraCli
}

func newTestUpgradeOperation(desiredVersion string, opts map[string]string) *cloudv1.UpgradeOperation {
	now := metav1.Now()
	return &cloudv1.UpgradeOperation{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "uo-test",
			Namespace: "default",
		},
		Status: cloudv1.UpgradeOperationStatus{
			FinishedSteps: cloudv1.NewSteps(),
			CurrentStep: &cloudv1.Step{
				Target:    cloudv1.CDCNodes,
				Version:   desiredVersion,
				Options:   opts,
				BeginTime: &now,
			},
		},
	}
}

func newTestCluster() *cloudv1.Cluster {
	return &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: testCDCClusterName,
		},
	}
}

func newTiCDCCluster(id, nodeVersion string, phase ticdcpb.TiCDCCluster_Phase) *ticdcpb.TiCDCCluster {
	return &ticdcpb.TiCDCCluster{
		Id:          id,
		NodeVersion: nodeVersion,
		Phase:       phase,
	}
}

func makeCDCPod(name, namespace, cdcInstance, k8sVersion string) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: namespace,
			Labels: map[string]string{
				"app.kubernetes.io/instance":  cdcInstance,
				"app.kubernetes.io/component": string(cloudv1.ComponentTiCDC),
			},
		},
		Spec: corev1.PodSpec{
			NodeName: "node-0",
			NodeSelector: map[string]string{
				consts.K8sVersionLabelKey: k8sVersion,
			},
		},
	}
}

func TestCDCUpgrader_Upgrade(t *testing.T) {
	listErr := errors.New("list failed")
	updateErr := errors.New("update failed")

	tests := []struct {
		name             string
		opts             map[string]string
		ticdcResp        *ticdcpb.ListTiCDCClustersResponse
		listErr          error
		updateErr        error
		dpObjects        []client.Object
		wantErr          bool
		wantErrContain   string
		wantRequeue      bool
		wantRequeueAfter time.Duration
		wantFinishStep   bool
		wantListFilter   []string
		wantUpdated      map[string]string
	}{
		{
			name:           "list TiCDC clusters error",
			listErr:        listErr,
			wantErr:        true,
			wantErrContain: "failed to list TiCDC clusters",
		},
		{
			name: "no clusters after filter finishes step",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "1.30", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantFinishStep: true,
		},
		{
			name: "skips DELETING clusters",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "1.26", ticdcpb.TiCDCCluster_DELETING),
				},
			},
			wantFinishStep: true,
		},
		{
			name: "target CDC annotation passed to list API",
			opts: map[string]string{TargetCDCUpgradeAnnotation: testCDCTiCDCID},
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "1.26", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantListFilter: []string{testCDCTiCDCID},
			wantRequeue:    true,
			wantUpdated:    map[string]string{testCDCTiCDCID: testCDCDesiredVer},
		},
		{
			name: "serial triggers node version update and requeues",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "1.26", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantRequeue: true,
			wantUpdated: map[string]string{testCDCTiCDCID: testCDCDesiredVer},
		},
		{
			name: "serial update API error",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "1.26", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			updateErr:      updateErr,
			wantErr:        true,
			wantErrContain: "failed to update TiCDC",
		},
		{
			name: "serial waits for pods to reschedule",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, testCDCDesiredVer, ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantRequeue: true,
		},
		{
			name: "serial waits for Running phase",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, testCDCDesiredVer, ticdcpb.TiCDCCluster_CREATING),
				},
			},
			dpObjects: []client.Object{
				makeCDCPod("cdc-pod-0", cloudv1.ClusterPrefixId(testCDCClusterName), testCDCTiCDCID, testCDCDesiredVer),
			},
			wantRequeueAfter: time.Minute,
		},
		{
			name: "serial completes when upgraded and pods ready",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, testCDCDesiredVer, ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			dpObjects: []client.Object{
				makeCDCPod("cdc-pod-0", cloudv1.ClusterPrefixId(testCDCClusterName), testCDCTiCDCID, testCDCDesiredVer),
			},
			wantFinishStep: true,
		},
		{
			name: "parallel requeues while upgrade in progress",
			opts: map[string]string{ParallelCDCUpgradeAnnotation: "true"},
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "1.26", ticdcpb.TiCDCCluster_RUNNING),
					newTiCDCCluster(testCDCTiCDCID2, "1.26", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantRequeueAfter: time.Minute,
			wantUpdated: map[string]string{
				testCDCTiCDCID:  testCDCDesiredVer,
				testCDCTiCDCID2: testCDCDesiredVer,
			},
		},
		{
			name: "parallel completes when all clusters ready",
			opts: map[string]string{ParallelCDCUpgradeAnnotation: "true"},
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, testCDCDesiredVer, ticdcpb.TiCDCCluster_RUNNING),
					newTiCDCCluster(testCDCTiCDCID2, testCDCDesiredVer, ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			dpObjects: []client.Object{
				makeCDCPod("cdc-pod-a", cloudv1.ClusterPrefixId(testCDCClusterName), testCDCTiCDCID, testCDCDesiredVer),
				makeCDCPod("cdc-pod-b", cloudv1.ClusterPrefixId(testCDCClusterName), testCDCTiCDCID2, testCDCDesiredVer),
			},
			wantFinishStep: true,
		},
		{
			name: "invalid node version still included for upgrade",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster(testCDCTiCDCID, "not-a-version", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantRequeue: true,
			wantUpdated: map[string]string{testCDCTiCDCID: testCDCDesiredVer},
		},
		{
			name: "only upgrades clusters below desired version",
			ticdcResp: &ticdcpb.ListTiCDCClustersResponse{
				TicdcClusters: []*ticdcpb.TiCDCCluster{
					newTiCDCCluster("cdc-high", "1.30", ticdcpb.TiCDCCluster_RUNNING),
					newTiCDCCluster(testCDCTiCDCID, "1.26", ticdcpb.TiCDCCluster_RUNNING),
				},
			},
			wantRequeue: true,
			wantUpdated: map[string]string{testCDCTiCDCID: testCDCDesiredVer},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			uo := newTestUpgradeOperation(testCDCDesiredVer, tt.opts)
			ticdcCli := &recordingTiCDCClient{
				listResp:  tt.ticdcResp,
				listErr:   tt.listErr,
				updateErr: tt.updateErr,
			}
			upgrader, infraCli := newTestCDCUpgrader(t, ticdcCli, uo)

			scheme := runtime.NewScheme()
			_ = corev1.AddToScheme(scheme)
			dpBuilder := fake.NewClientBuilder().WithScheme(scheme)
			if len(tt.dpObjects) > 0 {
				dpBuilder = dpBuilder.WithObjects(tt.dpObjects...)
			}
			env := &stubEnvProvider{dpCli: dpBuilder.Build()}

			result, err := upgrader.Upgrade(context.Background(), uo, env, newTestCluster())

			if tt.wantErr {
				if err == nil {
					t.Fatal("Upgrade() expected error, got nil")
				}
				if tt.wantErrContain != "" && !strings.Contains(err.Error(), tt.wantErrContain) {
					t.Fatalf("Upgrade() error = %q, want substring %q", err.Error(), tt.wantErrContain)
				}
			} else if err != nil {
				t.Fatalf("Upgrade() unexpected error: %v", err)
			}

			if tt.wantRequeue && !result.Requeue {
				t.Errorf("Upgrade() Requeue = false, want true")
			}
			if !tt.wantRequeue && result.Requeue {
				t.Errorf("Upgrade() Requeue = true, want false")
			}
			if tt.wantRequeueAfter != 0 && result.RequeueAfter != tt.wantRequeueAfter {
				t.Errorf("Upgrade() RequeueAfter = %v, want %v", result.RequeueAfter, tt.wantRequeueAfter)
			}
			if tt.wantRequeueAfter == 0 && result.RequeueAfter != 0 && !tt.wantFinishStep {
				t.Errorf("Upgrade() RequeueAfter = %v, want zero", result.RequeueAfter)
			}

			if tt.wantListFilter != nil {
				if fmt.Sprint(ticdcCli.lastTicdcFilter) != fmt.Sprint(tt.wantListFilter) {
					t.Errorf("ListTiCDCClusters filter = %v, want %v", ticdcCli.lastTicdcFilter, tt.wantListFilter)
				}
			}
			if ticdcCli.lastTidbClusterID != testCDCClusterName {
				t.Errorf("ListTiCDCClusters tidbClusterID = %q, want %q", ticdcCli.lastTidbClusterID, testCDCClusterName)
			}

			if tt.wantUpdated != nil {
				for id, ver := range tt.wantUpdated {
					if got := ticdcCli.updatedNodeVersion[id]; got != ver {
						t.Errorf("UpdateTiCDCNodeVersion(%s) = %q, want %q", id, got, ver)
					}
				}
			}

			gotUO := &cloudv1.UpgradeOperation{}
			if err := infraCli.Get(context.Background(), client.ObjectKeyFromObject(uo), gotUO); err != nil {
				t.Fatalf("get UpgradeOperation: %v", err)
			}
			if tt.wantFinishStep {
				if gotUO.Status.CurrentStep != nil {
					t.Fatal("expected CurrentStep to be cleared after FinishUpgrading")
				}
			} else if gotUO.Status.CurrentStep == nil {
				t.Fatal("expected CurrentStep to remain while upgrade is in progress")
			}
		})
	}
}