package auto

import (
	"context"
	"testing"

	"github.com/Masterminds/semver/v3"
	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	crrfMock "github.com/tidbcloud/dataflow-service/app/models/crrf/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repoMock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	mgmtMock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

// --- Helpers ---

func setupCRRFMinVer(t *testing.T, ver string) {
	c, err := semver.NewConstraint(">=" + ver)
	if err != nil {
		t.Fatal(err)
	}
	internal_config.Global().BackupConfig.CRRFEnableVersionConstraintStr = ver
	internal_config.Global().BackupConfig.CRRFEnableVersionConstraint = c
}

func newEnableExecCtx() run.ExecuteContext {
	ctx := fastflow_entity.CtxWithRunningTaskIns(context.Background(), &fastflow_entity.TaskInstance{DagInsID: "test-dag"})
	return run.NewDefExecuteContext(ctx, mapShareData{}, nil, nil, nil)
}

// newTriggeredExecCtx returns an ExecuteContext with autoCRRFEnableTriggeredKey pre-set,
// simulating the state after run() has called markAutoCRRFEnableTriggered.
func newTriggeredExecCtx() run.ExecuteContext {
	ctx := fastflow_entity.CtxWithRunningTaskIns(context.Background(), &fastflow_entity.TaskInstance{DagInsID: "test-dag"})
	sd := mapShareData{autoCRRFEnableTriggeredKey: "true"}
	return run.NewDefExecuteContext(ctx, sd, nil, nil, nil)
}

func awsBS(clusterID uint64, pitrOn, dualOn bool) *entity.BackupSetting {
	bs := &entity.BackupSetting{InfraProviderType: entity.InfraProviderTypeAWS}
	bs.ClusterID = entity.ID(clusterID)
	bs.ProjectID = entity.ID(1)
	bs.TenantID = entity.ID(1)
	if pitrOn {
		bs.PiTR = &entity.PiTRSetting{Enable: true}
	}
	if dualOn {
		bs.Bucket = &entity.BucketSetting{
			DualRegionEnable: true,
			Buckets: []entity.BackupSettingBucket{
				{Master: true},
				{Master: false, Regions: []string{"us-west-2"}},
			},
		}
	}
	return bs
}

func fakeCluster(ver string) *mgmt.Cluster {
	return &mgmt.Cluster{
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{Version: ver},
		},
	}
}

// --- Suite ---

type enableSuite struct {
	suite.Suite
	bsRepo     *repoMock.BackupSettingRepo
	mgmtCli    *mgmtMock.MgmtSvcGrpcClient
	crrfModel  *crrfMock.CRRFModel
	enableBase autoCRRFEnableBase
}

func TestAutoCRRFEnableSuite(t *testing.T) { suite.Run(t, new(enableSuite)) }

func (s *enableSuite) SetupTest() {
	s.bsRepo = repoMock.NewBackupSettingRepo(s.T())
	s.mgmtCli = mgmtMock.NewMgmtSvcGrpcClient(s.T())
	s.crrfModel = crrfMock.NewCRRFModel(s.T())
	s.enableBase = autoCRRFEnableBase{bsRepo: s.bsRepo, mgmtCli: s.mgmtCli, crrfModel: s.crrfModel}
	setupCRRFMinVer(s.T(), "v8.5.6")
	// Enable auto-CRRF feature flag globally for all tests; individual tests may override.
	internal_config.Global().BackupConfig.CRRFOverrides = internal_config.CRRFOverrides{
		{AutoCRRF: true, Gate: internal_config.FeatureGate{All: true}},
	}
}

func (s *enableSuite) TearDownTest() {
	s.bsRepo.AssertExpectations(s.T())
	s.mgmtCli.AssertExpectations(s.T())
	s.crrfModel.AssertExpectations(s.T())
}

// Cluster version too low → skip (nil error, no EnableCRRF call).
func (s *enableSuite) Test_VersionTooLow_Skip() {
	clusterID := uint64(1)
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(awsBS(clusterID, true, true), nil).Once()
	s.mgmtCli.On("GetAnyDedicatedClusterBasicByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.5"), nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err)
}

// PiTR not enabled → skip.
func (s *enableSuite) Test_PiTRNotEnabled_Skip() {
	clusterID := uint64(2)
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(awsBS(clusterID, false, true), nil).Once()
	s.mgmtCli.On("GetAnyDedicatedClusterBasicByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err)
}

// Dual-region not enabled → skip.
func (s *enableSuite) Test_DualRegionNotEnabled_Skip() {
	clusterID := uint64(3)
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(awsBS(clusterID, true, false), nil).Once()
	s.mgmtCli.On("GetAnyDedicatedClusterBasicByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err)
}

// CRRF already enabling → idempotent skip.
func (s *enableSuite) Test_CRRFAlreadyEnabling_Skip() {
	clusterID := uint64(4)
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(awsBS(clusterID, true, true), nil).Once()
	s.mgmtCli.On("GetAnyDedicatedClusterBasicByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusEnabling}, nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err)
}

// Non-AWS provider → skip.
func (s *enableSuite) Test_NonAWSProvider_Skip() {
	clusterID := uint64(17)
	bs := awsBS(clusterID, true, true)
	bs.InfraProviderType = entity.InfraProviderTypeGCP
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(bs, nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err)
}

// autoCRRF feature flag not set → skip.
func (s *enableSuite) Test_AutoCRRFFeatureFlagDisabled_Skip() {
	clusterID := uint64(19)
	internal_config.Global().BackupConfig.CRRFOverrides = nil
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(awsBS(clusterID, true, true), nil).Once()
	s.mgmtCli.On("GetAnyDedicatedClusterBasicByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err)
}

// Happy path: all conditions met → EnableCRRF called, nil error.
func (s *enableSuite) Test_HappyPath_EnableCRRFCalled() {
	clusterID := uint64(5)
	s.bsRepo.On("GetByFilter", mock.Anything, &backupRepo.BackupSettingFilter{ClusterID: &clusterID}).
		Return(awsBS(clusterID, true, true), nil).Once()
	s.mgmtCli.On("GetAnyDedicatedClusterBasicByID", mock.Anything, clusterID).Return(fakeCluster("v8.5.6"), nil).Once()
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabled}, nil).Once()
	s.crrfModel.On("EnableCRRF", mock.Anything, mock.MatchedBy(func(req *domain.EnableCRRFReq) bool {
		return req.ClusterID == clusterID && req.SecondaryRegion == "us-west-2"
	})).Return(nil).Once()

	err := s.enableBase.run(newEnableExecCtx(), clusterID)
	s.Nil(err) // nil → Wrapper proceeds to RunAfter
}

// RunAfter: CRRF enabled → nil (done, stops retry loop).
func (s *enableSuite) Test_RunAfter_Enabled_Nil() {
	clusterID := uint64(6)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusEnabled}, nil).Once()

	err := s.enableBase.poll(newTriggeredExecCtx(), clusterID)
	s.Nil(err)
}

// RunAfter: CRRF still enabling → retry.
func (s *enableSuite) Test_RunAfter_StillEnabling_Retry() {
	clusterID := uint64(7)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusEnabling}, nil).Once()

	err := s.enableBase.poll(newTriggeredExecCtx(), clusterID)
	s.NotNil(err)
}

// RunAfter: CRRF disabled with error → fatal error.
func (s *enableSuite) Test_RunAfter_DisabledWithError_FatalError() {
	clusterID := uint64(8)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabled, ErrorMessage: "some dag failure"}, nil).Once()

	err := s.enableBase.poll(newTriggeredExecCtx(), clusterID)
	s.NotNil(err)
	s.Contains(err.Error(), "CRRF enable DAG failed")
}

// RunAfter: CRRF disabled without error (skip/restart case) → nil.
func (s *enableSuite) Test_RunAfter_DisabledNoError_Nil() {
	clusterID := uint64(18)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabled}, nil).Once()

	err := s.enableBase.poll(newTriggeredExecCtx(), clusterID)
	s.Nil(err)
}

// RunAfter: when run didn't trigger enable, skip polling directly.
func (s *enableSuite) Test_RunAfter_NotTriggered_SkipPolling() {
	clusterID := uint64(20)

	err := s.enableBase.poll(newEnableExecCtx(), clusterID)
	s.Nil(err)
}