package task

import (
	"context"
	"reflect"
	"testing"
	"time"

	"github.com/go-logr/logr"
	"github.com/onsi/gomega"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"go.uber.org/mock/gomock"
	"gopkg.in/yaml.v3"

	k8sent "github.com/tidbcloud/devops/entity/k8supgrade"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/service"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/generator"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/infrainfo"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/task/repo"
)

var (
	deadline    = time.Date(2024, 12, 31, 12, 0, 0, 0, time.UTC)
	deadlineStr = deadline.Format(cloudv1.UpgradeOperationScheduleLayout)
)

const (
	testCRString = `name: upgrade-test-eks-uo
namespace: test
appboxName: eks-us-west-2
schedule: 2024-12-31 12:00:00 +0000 UTC
steps:
  - target: controlplane
    version: "1.30"
`
	testUOName      = "upgrade-test-eks-uo"
	testUONamespace = "test"
)

func buildTestUOFromCR(schedule string) *k8sent.UpgradeOperation {
	var uo k8sent.UpgradeOperation
	_ = yaml.Unmarshal([]byte(testCRString), &uo)
	if schedule != "" {
		uo.Schedule = schedule
	}
	uo.Annotations = map[string]string{}
	return &uo
}

func equalUpgradeOperation(got, want *k8sent.UpgradeOperation) bool {
	if got == nil || want == nil {
		return got == want
	}
	gotCopy, wantCopy := *got, *want
	gotCopy.Steps, wantCopy.Steps = nil, nil
	if !reflect.DeepEqual(gotCopy, wantCopy) {
		return false
	}
	if len(got.Steps) != len(want.Steps) {
		return false
	}
	for i := range got.Steps {
		if !reflect.DeepEqual(got.Steps[i], want.Steps[i]) {
			return false
		}
	}
	return true
}

func buildTestUO(name, ns, schedule string, status *k8sent.UpgradeOperationStatus) *k8sent.UpgradeOperation {
	return &k8sent.UpgradeOperation{
		Name:                  name,
		Namespace:             ns,
		Schedule:              schedule,
		AppboxName:            "eks-us-west-2",
		MaintenancePolicyMode: MaintenancePolicyModeLegacy,
		Annotations:           map[string]string{},
		Steps: []*k8sent.UpgradeStep{
			{
				Target:  string(cloudv1.ControlPlane),
				Version: "1.30",
			},
		},
		Status: status,
	}
}

func Test_svc_getUpgradeOperation(t *testing.T) {
	tests := []struct {
		name      string
		task      *k8sent.Task
		cluster   *k8sent.K8sCluster
		schedule  time.Time
		mockCalls func(*generator.MockAPI)
		expect    func(*gomega.WithT, *k8sent.K8sCluster)
		wantUo    *k8sent.UpgradeOperation
		wantErr   bool
	}{
		{
			name:    "first generation disables auto generation by default",
			task:    &k8sent.Task{Status: k8sent.TaskStatusPending},
			cluster: &k8sent.K8sCluster{Info: &k8sent.K8sInfo{}},
			mockCalls: func(mockGen *generator.MockAPI) {
				mockGen.EXPECT().GenerateUpgradeOperationKey(gomock.Any(), gomock.Any()).Return(testUOName, testUONamespace)
				mockGen.EXPECT().GenerateUpgradeOperation(context.TODO(), gomock.Any(), testUOName, testUONamespace).Return(buildTestUO(testUOName, testUONamespace, "", nil), "1.30")
			},
			expect: func(g *gomega.WithT, cluster *k8sent.K8sCluster) {
				g.Expect(cluster.DisableAutoGeneration).To(gomega.BeTrue())
				g.Expect(cluster.DesiredVersion).To(gomega.Equal("1.30"))
			},
			wantUo: buildTestUO(testUOName, testUONamespace, "", nil),
		},
		{
			name:     "with schedule",
			task:     &k8sent.Task{Status: k8sent.TaskStatusPending},
			cluster:  &k8sent.K8sCluster{Info: &k8sent.K8sInfo{}},
			schedule: deadline,
			mockCalls: func(mockGen *generator.MockAPI) {
				mockGen.EXPECT().GenerateUpgradeOperationKey(gomock.Any(), gomock.Any()).Return(testUOName, testUONamespace)
				mockGen.EXPECT().GenerateUpgradeOperation(context.TODO(), gomock.Any(), testUOName, testUONamespace).Return(buildTestUO(testUOName, testUONamespace, "", nil), "1.30")
			},
			expect: func(g *gomega.WithT, cluster *k8sent.K8sCluster) {
				g.Expect(cluster.DisableAutoGeneration).To(gomega.BeTrue())
				g.Expect(cluster.DesiredVersion).To(gomega.Equal("1.30"))
			},
			wantUo: buildTestUO(testUOName, testUONamespace, deadline.Format(cloudv1.UpgradeOperationScheduleLayout), nil),
		},
		{
			name:    "without schedule",
			task:    &k8sent.Task{Status: k8sent.TaskStatusPending},
			cluster: &k8sent.K8sCluster{Info: &k8sent.K8sInfo{}},
			mockCalls: func(mockGen *generator.MockAPI) {
				mockGen.EXPECT().GenerateUpgradeOperationKey(gomock.Any(), gomock.Any()).Return(testUOName, testUONamespace)
				mockGen.EXPECT().GenerateUpgradeOperation(context.TODO(), gomock.Any(), testUOName, testUONamespace).Return(buildTestUO(testUOName, testUONamespace, "", nil), "1.30")
			},
			expect: func(g *gomega.WithT, cluster *k8sent.K8sCluster) {
				g.Expect(cluster.DisableAutoGeneration).To(gomega.BeTrue())
				g.Expect(cluster.DesiredVersion).To(gomega.Equal("1.30"))
			},
			wantUo: buildTestUO(testUOName, testUONamespace, "", nil),
		},
		{
			name:     "disable auto generation",
			task:     &k8sent.Task{Status: k8sent.TaskStatusPending},
			cluster:  &k8sent.K8sCluster{Info: &k8sent.K8sInfo{}, DisableAutoGeneration: true, UpgradeOperationCR: testCRString},
			schedule: deadline,
			expect: func(g *gomega.WithT, cluster *k8sent.K8sCluster) {
				g.Expect(cluster.DesiredVersion).To(gomega.Equal("1.30"))
			},
			wantUo: buildTestUOFromCR(deadline.Format(cloudv1.UpgradeOperationScheduleLayout)),
		},
		{
			name:    "re-enabled auto generation keeps it enabled",
			task:    &k8sent.Task{Status: k8sent.TaskStatusPending},
			cluster: &k8sent.K8sCluster{Info: &k8sent.K8sInfo{}, DisableAutoGeneration: false, UpgradeOperationCR: testCRString},
			mockCalls: func(mockGen *generator.MockAPI) {
				mockGen.EXPECT().GenerateUpgradeOperationKey(gomock.Any(), gomock.Any()).Return(testUOName, testUONamespace)
				mockGen.EXPECT().GenerateUpgradeOperation(context.TODO(), gomock.Any(), testUOName, testUONamespace).Return(buildTestUO(testUOName, testUONamespace, "", nil), "1.30")
			},
			expect: func(g *gomega.WithT, cluster *k8sent.K8sCluster) {
				g.Expect(cluster.DisableAutoGeneration).To(gomega.BeFalse())
				g.Expect(cluster.DesiredVersion).To(gomega.Equal("1.30"))
			},
			wantUo: buildTestUO(testUOName, testUONamespace, "", nil),
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			mockGenerator := generator.NewMockAPI(ctrl)
			if tt.mockCalls != nil {
				tt.mockCalls(mockGenerator)
			}

			s := &svc{
				generator: mockGenerator,
			}
			gotUo, err := s.getUpgradeOperation(context.TODO(), k8sent.NewReadOnlyTask(tt.task), tt.cluster, tt.schedule)
			if (err != nil) != tt.wantErr {
				t.Errorf("getUpgradeOperation() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if !equalUpgradeOperation(gotUo, tt.wantUo) {
				t.Errorf("getUpgradeOperation() gotUo = %v, want %v", gotUo, tt.wantUo)
			}

			g := gomega.NewGomegaWithT(t)
			if tt.expect != nil {
				tt.expect(g, tt.cluster)
			}
			gotUoBytes, err := yaml.Marshal(gotUo)
			g.Expect(err).To(gomega.BeNil())
			g.Expect(string(gotUoBytes)).To(gomega.Equal(tt.cluster.UpgradeOperationCR))
		})
	}
}

// buildSyncPendingSvc builds a minimal svc wired for syncPendingTask tests.
func buildSyncPendingSvc(
	taskRepo repo.Repo,
	infraInfoSvc infrainfo.ServiceAPI,
	gen generator.API,
) *svc {
	return &svc{
		BaseService:   service.NewBaseService(svcName),
		taskRepo:      taskRepo,
		infraInfoSvc:  infraInfoSvc,
		generator:     gen,
		syncTasksChan: make(chan uint64, defaultSyncChanSize),
	}
}

func Test_syncPendingTask_versionConflict_triggersResync(t *testing.T) {
	g := gomega.NewGomegaWithT(t)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	const (
		taskID      = uint64(42)
		clusterName = "eks-us-west-2"
		clusterID   = uint64(1)
		version     = uint64(5)
	)

	existingCluster := &k8sent.K8sCluster{
		ID:            clusterID,
		Name:          clusterName,
		TaskID:        taskID,
		CloudProvider: "aws",
		Region:        "us-west-2",
		Version:       version,
	}
	task := &k8sent.Task{
		ID:                  taskID,
		Status:              k8sent.TaskStatusPending,
		CloudProvider:       "aws",
		VersionToBeUpgraded: "1.27",
		Deadline:            deadline.Unix(),
		K8sClusterList:      []*k8sent.K8sCluster{existingCluster},
		ExcludedK8sClusters: []*k8sent.K8sCluster{},
		SelectorType:        k8sent.SelectorTypeNone,
	}

	detail := &k8sent.K8sInfo{
		CloudProvider:  "aws",
		Region:         "us-west-2",
		K8sName:        clusterName,
		BusinessRegion: "us-west",
		AppboxName:     clusterName,
	}

	mockRepo := repo.NewMockRepo(ctrl)
	mockInfra := infrainfo.NewMockServiceAPI(ctrl)
	mockGen := generator.NewMockAPI(ctrl)

	mockRepo.EXPECT().GetTask(gomock.Any(), taskID, true).Return(task, nil)
	mockInfra.EXPECT().ListK8SInformation(gomock.Any(), &infrainfo.ListK8SInformationReq{
		CloudProvider: "aws",
		Version:       "1.27",
	}).Return(k8sent.K8sInfoList{
		Items: []*k8sent.K8sInfo{{CloudProvider: "aws", Region: "us-west-2", K8sName: clusterName}},
	}, nil)
	mockInfra.EXPECT().GetK8sInformation(gomock.Any(), gomock.Any()).Return(detail, nil)
	mockGen.EXPECT().GenerateUpgradeOperationKey(gomock.Any(), gomock.Any()).Return("uo-name", "uo-ns")
	mockGen.EXPECT().GenerateUpgradeOperation(gomock.Any(), gomock.Any(), "uo-name", "uo-ns").
		Return(buildTestUO("uo-name", "uo-ns", "", nil), "1.30")
	// Simulate a concurrent write: version guard fails for the cluster.
	mockRepo.EXPECT().UpsertK8sClustersWithVersionGuard(gomock.Any(), gomock.Any()).
		Return([]string{clusterName}, nil)
	mockRepo.EXPECT().PatchTask(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)

	s := buildSyncPendingSvc(mockRepo, mockInfra, mockGen)
	ctx := log.NewContext(context.Background(), logr.Discard())
	s.syncPendingTask(ctx, taskID)

	// Version conflict must have triggered a resync.
	g.Expect(s.syncTasksChan).To(gomega.HaveLen(1))
	resyncID := <-s.syncTasksChan
	g.Expect(resyncID).To(gomega.Equal(taskID))
}

func Test_syncPendingTask_noConflict_noResync(t *testing.T) {
	g := gomega.NewGomegaWithT(t)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	const (
		taskID      = uint64(43)
		clusterName = "eks-us-east-1"
		clusterID   = uint64(2)
		version     = uint64(3)
	)

	existingCluster := &k8sent.K8sCluster{
		ID:            clusterID,
		Name:          clusterName,
		TaskID:        taskID,
		CloudProvider: "aws",
		Region:        "us-east-1",
		Version:       version,
	}
	task := &k8sent.Task{
		ID:                  taskID,
		Status:              k8sent.TaskStatusPending,
		CloudProvider:       "aws",
		VersionToBeUpgraded: "1.27",
		Deadline:            deadline.Unix(),
		K8sClusterList:      []*k8sent.K8sCluster{existingCluster},
		ExcludedK8sClusters: []*k8sent.K8sCluster{},
		SelectorType:        k8sent.SelectorTypeNone,
	}

	detail := &k8sent.K8sInfo{
		CloudProvider:  "aws",
		Region:         "us-east-1",
		K8sName:        clusterName,
		BusinessRegion: "us-east",
		AppboxName:     clusterName,
	}

	mockRepo := repo.NewMockRepo(ctrl)
	mockInfra := infrainfo.NewMockServiceAPI(ctrl)
	mockGen := generator.NewMockAPI(ctrl)

	mockRepo.EXPECT().GetTask(gomock.Any(), taskID, true).Return(task, nil)
	mockInfra.EXPECT().ListK8SInformation(gomock.Any(), gomock.Any()).Return(k8sent.K8sInfoList{
		Items: []*k8sent.K8sInfo{{CloudProvider: "aws", Region: "us-east-1", K8sName: clusterName}},
	}, nil)
	mockInfra.EXPECT().GetK8sInformation(gomock.Any(), gomock.Any()).Return(detail, nil)
	mockGen.EXPECT().GenerateUpgradeOperationKey(gomock.Any(), gomock.Any()).Return("uo-name", "uo-ns")
	mockGen.EXPECT().GenerateUpgradeOperation(gomock.Any(), gomock.Any(), "uo-name", "uo-ns").
		Return(buildTestUO("uo-name", "uo-ns", "", nil), "1.30")
	// No conflicts: all updates succeed.
	mockRepo.EXPECT().UpsertK8sClustersWithVersionGuard(gomock.Any(), gomock.Any()).
		Return(nil, nil)
	mockRepo.EXPECT().PatchTask(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)

	s := buildSyncPendingSvc(mockRepo, mockInfra, mockGen)
	ctx := log.NewContext(context.Background(), logr.Discard())
	s.syncPendingTask(ctx, taskID)

	// No conflict: channel must remain empty.
	g.Expect(s.syncTasksChan).To(gomega.HaveLen(0))
}

func Test_handleTask_inflightDeduplication(t *testing.T) {
	g := gomega.NewGomegaWithT(t)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	const taskID = uint64(99)

	mockRepo := repo.NewMockRepo(ctrl)
	// GetTask must NOT be called at all because the task is already in-flight.
	mockRepo.EXPECT().GetTask(gomock.Any(), gomock.Any(), gomock.Any()).Times(0)

	s := buildSyncPendingSvc(mockRepo, nil, nil)
	// Pre-mark the task as in-flight, simulating a concurrent worker.
	s.inflightTasks.Store(taskID, struct{}{})

	ctx := log.NewContext(context.Background(), logr.Discard())
	s.handleTask(ctx, taskID)

	// The in-flight entry must still be present (we didn't call defer Delete).
	_, loaded := s.inflightTasks.Load(taskID)
	g.Expect(loaded).To(gomega.BeTrue())
}