package auto

import (
	"context"
	"testing"

	retry "github.com/avast/retry-go/v4"
	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"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
)

// mapShareData is a simple in-memory ShareDataOperator for tests.
type mapShareData map[string]string

func (m mapShareData) Get(key string) (string, bool) { v, ok := m[key]; return v, ok }
func (m mapShareData) Set(key string, val string)    { m[key] = val }

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

type disableSuite struct {
	suite.Suite
	crrfModel   *crrfMock.CRRFModel
	disableBase autoCRRFDisableBase
}

func TestAutoCRRFDisableSuite(t *testing.T) { suite.Run(t, new(disableSuite)) }

func (s *disableSuite) SetupTest() {
	s.crrfModel = crrfMock.NewCRRFModel(s.T())
	s.disableBase = autoCRRFDisableBase{crrfModel: s.crrfModel}
}

func (s *disableSuite) TearDownTest() {
	s.crrfModel.AssertExpectations(s.T())
}

// CRRF already disabled → nil error (idempotent).
func (s *disableSuite) Test_AlreadyDisabled_Skip() {
	clusterID := uint64(10)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabled}, nil).Once()

	err := s.disableBase.run(newDisableExecCtx(), clusterID)
	s.Nil(err)
}

// CRRF already disabling → nil (RunAfter will poll for completion).
func (s *disableSuite) Test_Disabling_NotSkipped_Nil() {
	clusterID := uint64(11)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabling}, nil).Once()

	err := s.disableBase.run(newDisableExecCtx(), clusterID)
	s.Nil(err)
}

// CRRF enabled → DisableCRRF called, nil error.
func (s *disableSuite) Test_Enabled_DisableCRRFCalled() {
	clusterID := uint64(12)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusEnabled}, nil).Once()
	s.crrfModel.On("DisableCRRF", mock.Anything, mock.MatchedBy(func(req *domain.DisableCRRFReq) bool {
		return req.ClusterID == clusterID
	})).Return(nil).Once()

	err := s.disableBase.run(newDisableExecCtx(), clusterID)
	s.Nil(err)
}

// CRRF enabling → DisableCRRF called.
func (s *disableSuite) Test_Enabling_DisableCRRFCalled() {
	clusterID := uint64(13)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusEnabling}, nil).Once()
	s.crrfModel.On("DisableCRRF", mock.Anything, mock.MatchedBy(func(req *domain.DisableCRRFReq) bool {
		return req.ClusterID == clusterID
	})).Return(nil).Once()

	err := s.disableBase.run(newDisableExecCtx(), clusterID)
	s.Nil(err)
}

// RunAfter: CRRF disabled → nil (done, stops retry loop).
func (s *disableSuite) Test_RunAfter_Disabled_Nil() {
	clusterID := uint64(14)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabled}, nil).Once()

	err := s.disableBase.poll(newDisableExecCtx(), clusterID)
	s.Nil(err)
}

// RunAfter: CRRF still disabling → retry.
func (s *disableSuite) Test_RunAfter_StillDisabling_Retry() {
	clusterID := uint64(15)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusDisabling}, nil).Once()

	err := s.disableBase.poll(newDisableExecCtx(), clusterID)
	s.NotNil(err)
	// verify it's a transient error, not an action sentinel
	s.False(retry.IsRecoverable(err) == false, "disabling status should be a recoverable error")
}

// RunAfter: CRRF disable DAG reached failed state → Unrecoverable error (stops infinite retry).
func (s *disableSuite) Test_RunAfter_Failed_Unrecoverable() {
	clusterID := uint64(16)
	s.crrfModel.On("QueryCRRF", mock.Anything, clusterID).
		Return(&domain.CRRFInfo{Status: entity.CRRFSettingStatusFailed, ErrorMessage: "dag startup failed"}, nil).Once()

	err := s.disableBase.poll(newDisableExecCtx(), clusterID)
	s.NotNil(err)
	s.False(retry.IsRecoverable(err), "expected unrecoverable error for failed CRRF status")
}