package cron

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	servermocks "github.com/tidbcloud/cluster-service-ng/domain/server/mocks"
	gatewaymocks "github.com/tidbcloud/cluster-service-ng/pkg/clustergateway/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"go.opentelemetry.io/otel"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type dmWorkerAutoscalerTestDeps struct {
	dmClusterRepo  *repomocks.DMClusterRepo
	sharedPoolRepo *repomocks.SharedPoolRepo
	lockRepo       *repomocks.LockRepo
	ngService      *servermocks.Service
	clusterGateway *gatewaymocks.Client
}

func newDMWorkerAutoscalerTestJob(t *testing.T) (*DMWorkerAutoscalerJob, dmWorkerAutoscalerTestDeps) {
	t.Helper()

	prevTracerProvider := otel.GetTracerProvider()
	tracerProvider := sdktrace.NewTracerProvider()
	otel.SetTracerProvider(tracerProvider)
	t.Cleanup(func() {
		_ = tracerProvider.Shutdown(context.Background())
		otel.SetTracerProvider(prevTracerProvider)
	})

	deps := dmWorkerAutoscalerTestDeps{
		dmClusterRepo:  repomocks.NewDMClusterRepo(t),
		sharedPoolRepo: repomocks.NewSharedPoolRepo(t),
		lockRepo:       repomocks.NewLockRepo(t),
		ngService:      servermocks.NewService(t),
		clusterGateway: gatewaymocks.NewClient(t),
	}
	return NewDMWorkerAutoscalerJob(
		deps.dmClusterRepo,
		deps.sharedPoolRepo,
		deps.lockRepo,
		deps.ngService,
		deps.clusterGateway,
		"*/30 * * * * *",
		5*time.Minute,
	), deps
}

func TestDMWorkerAutoscalerScaleUpImmediately(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	dmCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 1)
	sharedPool := testDMWorkerAutoscalerSharedPool(dmCluster.GetSharedPoolId())

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(3, nil).Once()
	expectDMWorkerScale(t, deps, dmCluster, 3)

	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))
}

func TestDMWorkerAutoscalerDoesNotBypassScaleDownStabilityWithStaleSnapshot(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	staleCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 1)
	latestCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 5)
	sharedPool := testDMWorkerAutoscalerSharedPool(staleCluster.GetSharedPoolId())

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, staleCluster.GetId()).Return(3, nil).Once()
	deps.lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	deps.ngService.EXPECT().
		GetDMCluster(mock.Anything, sharedPool.GetId(), staleCluster.GetId()).
		Return(latestCluster, nil).
		Once()

	require.NoError(t, job.processDMCluster(context.Background(), staleCluster))
}

func TestDMWorkerAutoscalerScaleDownAfterStableTarget(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	dmCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 3)
	sharedPool := testDMWorkerAutoscalerSharedPool(dmCluster.GetSharedPoolId())
	now := time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC)
	job.now = func() time.Time { return now }

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Times(4)
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(2, nil).Once()
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(0, nil).Times(3)

	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))

	now = now.Add(4 * time.Minute)
	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))

	now = now.Add(4 * time.Minute)
	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))

	expectDMWorkerScale(t, deps, dmCluster, 0)
	now = now.Add(time.Minute)
	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))
}

func TestDMWorkerAutoscalerSkipsWhenLockUnavailable(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	dmCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 1)
	sharedPool := testDMWorkerAutoscalerSharedPool(dmCluster.GetSharedPoolId())

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(2, nil).Once()
	deps.lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(nil, status.Error(codes.FailedPrecondition, "lock held")).Once()

	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))
}

func TestDMWorkerAutoscalerReturnsInternalLockError(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	dmCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 1)
	sharedPool := testDMWorkerAutoscalerSharedPool(dmCluster.GetSharedPoolId())

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(2, nil).Once()
	deps.lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(nil, status.Error(codes.Internal, "db unavailable")).Once()

	err := job.processDMCluster(context.Background(), dmCluster)
	require.Error(t, err)
	assert.Equal(t, codes.Internal, status.Code(err))
	assert.Contains(t, err.Error(), "acquire DM worker autoscaler lock")
}

func TestDMWorkerAutoscalerRetainsStableTargetWhenLockUnavailable(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	dmCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 3)
	sharedPool := testDMWorkerAutoscalerSharedPool(dmCluster.GetSharedPoolId())
	now := time.Date(2026, 6, 15, 0, 0, 0, 0, time.UTC)
	job.now = func() time.Time { return now }

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Times(3)
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(1, nil).Times(3)
	deps.lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(nil, status.Error(codes.FailedPrecondition, "lock held")).Once()

	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))
	now = now.Add(5 * time.Minute)
	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))

	expectDMWorkerScale(t, deps, dmCluster, 1)
	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))
}

func TestDMWorkerAutoscalerSkipsClusterWithRunningOperation(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	dmCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_AVAILABLE, 1)
	sharedPool := testDMWorkerAutoscalerSharedPool(dmCluster.GetSharedPoolId())
	modifyingCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_MODIFYING, 1)

	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Once()
	deps.clusterGateway.EXPECT().ListActiveDMSourceCount(mock.Anything, sharedPool, dmCluster.GetId()).Return(2, nil).Once()
	deps.lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	deps.ngService.EXPECT().GetDMCluster(mock.Anything, sharedPool.GetId(), dmCluster.GetId()).Return(modifyingCluster, nil).Once()

	require.NoError(t, job.processDMCluster(context.Background(), dmCluster))
}

func TestDMWorkerAutoscalerExecuteSkipsUnavailableClusters(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	unavailableCluster := testDMWorkerAutoscalerCluster("dm-1", apiv1.DMCluster_MODIFYING, 1)

	deps.dmClusterRepo.EXPECT().
		List(mock.Anything, mock.MatchedBy(func(filter *repo.DMClusterListFilter) bool {
			return filter != nil
		})).
		Return([]*entity.DMCluster{unavailableCluster}, nil).
		Once()

	require.NoError(t, job.Execute(context.Background()))
}

func TestDMWorkerAutoscalerContinuesAfterClusterFailure(t *testing.T) {
	job, deps := newDMWorkerAutoscalerTestJob(t)
	failedCluster := testDMWorkerAutoscalerCluster("dm-failed", apiv1.DMCluster_AVAILABLE, 1)
	healthyCluster := testDMWorkerAutoscalerCluster("dm-healthy", apiv1.DMCluster_AVAILABLE, 1)
	sharedPool := testDMWorkerAutoscalerSharedPool(failedCluster.GetSharedPoolId())

	deps.dmClusterRepo.EXPECT().
		List(mock.Anything, mock.Anything).
		Return([]*entity.DMCluster{failedCluster, healthyCluster}, nil).
		Once()
	deps.sharedPoolRepo.EXPECT().Load(mock.Anything, sharedPool.GetId(), false).Return(sharedPool, nil).Times(2)
	deps.clusterGateway.EXPECT().
		ListActiveDMSourceCount(mock.Anything, sharedPool, failedCluster.GetId()).
		Return(0, errors.New("gateway unavailable")).
		Once()
	deps.clusterGateway.EXPECT().
		ListActiveDMSourceCount(mock.Anything, sharedPool, healthyCluster.GetId()).
		Return(1, nil).
		Once()

	err := job.Execute(context.Background())
	require.Error(t, err)
	assert.Contains(t, err.Error(), failedCluster.GetId())
}

func expectDMWorkerScale(
	t *testing.T,
	deps dmWorkerAutoscalerTestDeps,
	dmCluster *entity.DMCluster,
	target uint32,
) {
	t.Helper()

	deps.lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	deps.ngService.EXPECT().
		GetDMCluster(mock.Anything, dmCluster.GetSharedPoolId(), dmCluster.GetId()).
		Return(dmCluster, nil).
		Once()
	deps.ngService.EXPECT().
		ScaleDMCluster(
			mock.MatchedBy(func(ctx context.Context) bool {
				return trace.SpanContextFromContext(ctx).HasTraceID()
			}),
			dmCluster.GetSharedPoolId(),
			dmCluster.GetId(),
			(*entity.NodeQuota)(nil),
			mock.MatchedBy(func(quota *entity.NodeQuota) bool {
				return quota != nil && quota.GetNodeCount() == target
			}),
			false,
		).
		Return(dmCluster, nil).
		Once()
}

func testDMWorkerAutoscalerCluster(id string, phase apiv1.DMCluster_Phase, workerCount uint32) *entity.DMCluster {
	return &entity.DMCluster{
		Id:           id,
		SharedPoolId: "shared-pool-1",
		Phase:        phase,
		Backend:      entity.DMBackendOperatorV2Proto,
		WorkerNodeQuota: &entity.NodeQuota{
			NodeCount: workerCount,
		},
	}
}

func testDMWorkerAutoscalerSharedPool(id string) *entity.SharedPool {
	return &entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: id},
	}
}

func TestDMWorkerAutoscalerJobMetadata(t *testing.T) {
	job, _ := newDMWorkerAutoscalerTestJob(t)
	assert.Equal(t, "DMWorkerAutoscalerJob", job.Name())
	assert.Equal(t, "*/30 * * * * *", job.CronExpression())
}