package service

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	ffentity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade/mock_facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
	"gorm.io/gorm"
)

func TestScaleChangefeedNewArchSuccess(t *testing.T) {
	ctrl := gomock.NewController(t)
	changefeedRepo := mock_repo.NewMockChangefeedRepo(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	workflow := &fakeWorkflow{}

	const (
		clusterID  = "10194411360446690844"
		feedID     = "cf-local"
		provider   = "alicloud"
		region     = "ap-southeast-1"
		kubernetes = "dev-ap-southeast-1-f01"
		keyspace   = "mVPMqMuJovrSVHt"
	)

	clusterName := buildEssentialTiCDCClusterName(&newArchShard{
		Provider:   provider,
		Region:     region,
		Kubernetes: kubernetes,
		ShareScope: "global",
		ShareKey:   "global",
	})
	var gotMethod string
	var gotPath string
	var gotKeyspace string
	var gotState string
	var encodeErr error
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		gotMethod = r.Method
		gotPath = r.URL.Path
		gotKeyspace = r.URL.Query().Get("keyspace")
		gotState = r.URL.Query().Get("state")
		w.Header().Set("Content-Type", "application/json")
		encodeErr = json.NewEncoder(w).Encode(&clusterproxy.ListResponse[clusterproxy.ChangefeedCommonInfo]{
			Total: 1,
			Items: []clusterproxy.ChangefeedCommonInfo{{
				ID:            feedID,
				FeedState:     "normal",
				CheckpointTSO: 4321,
			}},
		})
	}))
	defer srv.Close()

	setConfigForTest(t, &config.CentralConfig{
		ServerConfig: config.ServerConfig{Stage: config.StageDev},
		WorkerConfig: config.WorkerConfig{
			RegionalConfig: []config.RegionalConfig{{
				Provider: provider,
				Region:   region,
				K8sConfig: []config.K8sConfig{{
					Kubernetes:      kubernetes,
					ClusterProxyUrl: srv.URL,
				}},
			}},
		},
	})

	changefeed := &entity.CDCChangefeed{
		ID:            feedID,
		ClusterID:     clusterID,
		State:         constant.ChangfeedStateRunning,
		Arch:          constant.ChangefeedArchTiCDCNewArch,
		RCU:           8,
		CloudProvider: provider,
		Region:        region,
		Kubernetes:    kubernetes,
		Name:          "test-feed",
		CreatedAt:     time.Unix(1716870000, 0),
	}

	changefeedRepo.EXPECT().
		GetById(gomock.Any(), clusterID, feedID).
		Return(changefeed, nil)
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(10194411360446690844), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(&tidbmgmtservicev1.Cluster{
			ClusterId: 10194411360446690844,
			ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
				ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
					Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
						UserPrefix: keyspace,
						KeyspaceId: 554085,
					},
				},
			},
		}, nil)
	changefeedRepo.EXPECT().
		SyncState(gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, got *entity.CDCChangefeed) bizerr.BizError {
			require.Equal(t, constant.ChangfeedStateRunning, got.State)
			require.Equal(t, uint64(4321), got.CheckpointTSO)
			return nil
		})
	changefeedRepo.EXPECT().
		Save(gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, got *entity.CDCChangefeed) (*entity.CDCChangefeed, bizerr.BizError) {
			require.Equal(t, uint32(16), got.RCU)
			require.Equal(t, constant.ChangefeedStateScaling, got.State)
			return got, nil
		})

	svc := &cdcService{
		transProvider:  fakeTransProvider{},
		mgmtSvc:        mgmtSvc,
		changefeedRepo: changefeedRepo,
		workflow:       workflow,
	}

	got, err := svc.ScaleChangefeed(context.Background(), clusterID, feedID, 16)

	require.NoError(t, err)
	require.NotNil(t, got)
	require.NoError(t, encodeErr)
	require.Equal(t, http.MethodGet, gotMethod)
	require.Equal(
		t,
		"/proxy/namespace/tidb-serverless/cluster/serverless-cluster/ticdc/"+clusterName+"/api/v2/changefeeds",
		gotPath,
	)
	require.Equal(t, keyspace, gotKeyspace)
	require.Equal(t, "all", gotState)
	require.Equal(t, constant.ChangefeedStateScaling, got.State)
	require.Equal(t, uint32(16), got.RCU)
	require.Equal(t, appworkflow.DagScaleNewArchCluster, workflow.dagID)
	require.Equal(t, clusterID, workflow.vars["cluster_id"])
	require.Equal(t, feedID, workflow.vars["changefeed_id"])
	require.Equal(t, provider, workflow.tags[appworkflow.TagProvider])
	require.Equal(t, region, workflow.tags[appworkflow.TagRegion])
	require.Equal(t, kubernetes, workflow.tags[appworkflow.TagKubernetes])
}

func TestScaleChangefeedRejectsLegacyArch(t *testing.T) {
	ctrl := gomock.NewController(t)
	changefeedRepo := mock_repo.NewMockChangefeedRepo(ctrl)

	changefeedRepo.EXPECT().
		GetById(gomock.Any(), "cluster-1", "cf-1").
		Return(&entity.CDCChangefeed{
			ID:        "cf-1",
			ClusterID: "cluster-1",
			Arch:      constant.ChangefeedArchReplicationWorker,
		}, nil)

	svc := &cdcService{changefeedRepo: changefeedRepo}

	got, err := svc.ScaleChangefeed(context.Background(), "cluster-1", "cf-1", 4)

	require.Nil(t, got)
	require.Error(t, err)
	require.Contains(t, err.Error(), "only supported for ticdc newarch")
}

func TestScaleChangefeedRejectsNonRunningState(t *testing.T) {
	ctrl := gomock.NewController(t)
	changefeedRepo := mock_repo.NewMockChangefeedRepo(ctrl)

	changefeedRepo.EXPECT().
		GetById(gomock.Any(), "cluster-1", "cf-1").
		Return(&entity.CDCChangefeed{
			ID:        "cf-1",
			ClusterID: "cluster-1",
			Arch:      constant.ChangefeedArchTiCDCNewArch,
			State:     constant.ChangefeedStateScaling,
			RCU:       4,
		}, nil)

	svc := &cdcService{changefeedRepo: changefeedRepo}

	got, err := svc.ScaleChangefeed(context.Background(), "cluster-1", "cf-1", 8)

	require.Nil(t, got)
	require.Error(t, err)
	require.Contains(t, err.Error(), "only allowed to scale in running state")
}

type fakeTransProvider struct{}

func (fakeTransProvider) ReadDB(context.Context) *gorm.DB { return nil }

func (fakeTransProvider) WriteDB(context.Context) *gorm.DB { return nil }

func (fakeTransProvider) Transaction(ctx context.Context, cb func(ctx context.Context) error) error {
	return cb(ctx)
}

type fakeWorkflow struct {
	dagID string
	vars  map[string]string
	tags  map[string]string
}

func (f *fakeWorkflow) Close() {}

func (f *fakeWorkflow) RegisterDag(*ffentity.Dag, []run.Action) error { return nil }

func (f *fakeWorkflow) RunDagWithTags(
	dagID string,
	vars map[string]string,
	tags map[string]string,
) (*ffentity.DagInstance, error) {
	f.dagID = dagID
	f.vars = vars
	f.tags = tags
	return &ffentity.DagInstance{}, nil
}

func (f *fakeWorkflow) RetryDagIns(string, ...mod.CommandOptSetter) error { return nil }

func (f *fakeWorkflow) CancelDagIns(string, ...mod.CommandOptSetter) error { return nil }

func (f *fakeWorkflow) ListDagIns(*mod.ListDagInstanceInput) ([]*ffentity.DagInstance, error) {
	return nil, nil
}

func (f *fakeWorkflow) StartExecutor() error { return nil }