package cluster

import (
	"context"
	"errors"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	accountsdk "github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/fastflow/pkg/entity"
	"go.uber.org/mock/gomock"
	grpcmetadata "google.golang.org/grpc/metadata"
	"google.golang.org/protobuf/proto"

	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/app/bff/orchestrator"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account/mock_account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/infra/repository/db"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/infra/workflowimpl"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	workflowutil "github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func TestOpenAPIAction(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockAccountCli := mock_account.NewMockClient(ctrl)
	mockAccountCli.EXPECT().
		GetProject(gomock.Any(), gomock.Any()).
		DoAndReturn(func(ctx context.Context, projectId uint64) (*accountsdk.Project, error) {
			return &accountsdk.Project{
				Id:       projectId,
				TenantId: 1, // org id
			}, nil
		}).
		AnyTimes()
	mockAccountCli.EXPECT().
		GetTenantThirdPartyAccountProviders(gomock.Any(), gomock.Any()).
		DoAndReturn(func(ctx context.Context, id uint64) ([]openapicommonv1beta1.Region_CloudProvider, error) {
			return []openapicommonv1beta1.Region_CloudProvider{}, nil
		}).
		AnyTimes()
	mockAccountCli.EXPECT().
		UpsertClusterRelations(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(nil).
		AnyTimes()

	openAPIDedicatedClusterServer, openAPIOrchestrator, err := initService(mockAccountCli)
	require.NoError(t, err)

	// Create a cluster first
	ctx := grpcmetadata.NewIncomingContext(t.Context(), grpcmetadata.New(map[string]string{
		metadata.UserIDHeaderKey:           "1",
		metadata.OrgIDHeaderKey:            "1",
		metadata.DefaultProjectIDHeaderKey: "1",
		metadata.ProjectListHeaderKey:      "1,2,3",
		metadata.IsCheckPermissionKey:      "true",
	}))

	req := &openapidedicatedv1beta1.CreateClusterRequest{
		Cluster: &openapidedicatedv1beta1.Cluster{
			DisplayName: "get-test-1",
			RegionId:    "aws-us-west-2",
			TidbNodeSetting: &openapidedicatedv1beta1.Cluster_TidbNodeSetting{
				NodeSpecKey: "8C16G",
				TidbNodeGroups: []*openapidedicatedv1beta1.TidbNodeGroup{
					{
						NodeCount: 1,
					},
				},
			},
			TikvNodeSetting: &openapidedicatedv1beta1.Cluster_StorageNodeSetting{
				NodeSpecKey:   "8C16G",
				NodeCount:     3,
				StorageSizeGi: 50,
				StorageType:   openapidedicatedv1beta1.Cluster_StorageNodeSetting_Standard,
			},
			Labels: map[string]string{
				common.ProjectIDLabelKey: "1",
			},
		},
	}
	c, err := openAPIDedicatedClusterServer.CreateCluster(ctx, req)
	require.NoError(t, err)
	time.Sleep(3 * time.Second) // wait for the workflow dag to finish
	clusterID := c.ClusterId

	t.Run("scale", func(t *testing.T) {
		requests := []*orchestrator.OpenAPIRequest{
			{
				ActionName: OpenAPIScaleAction,
				Message: &openapidedicatedv1beta1.UpdateClusterRequest{
					Cluster: &openapidedicatedv1beta1.UpdateClusterRequest_Cluster{
						ClusterId: clusterID,
						TidbNodeSetting: &openapidedicatedv1beta1.UpdateClusterRequest_TidbNodeSetting{
							NodeSpecKey: "8C16G",
							TidbNodeGroups: []*openapidedicatedv1beta1.UpdateClusterRequest_TidbNodeSetting_TidbNodeGroup{
								{
									NodeCount: lo.ToPtr(int32(3)),
								},
							},
						},
						TikvNodeSetting: &openapidedicatedv1beta1.UpdateClusterRequest_StorageNodeSetting{
							NodeSpecKey:   "8C16G",
							NodeCount:     lo.ToPtr(int32(9)),
							StorageSizeGi: 500,
						},
					},
				},
			},
		}

		err := openAPIOrchestrator.ValidateOneByOne(ctx, requests)
		require.NoError(t, err)

		for _, item := range requests {
			validateOnly, err := getValidateOnlyField(item.Message)
			require.NoError(t, err)
			assert.True(t, validateOnly)
		}

		cluster, err := openAPIDedicatedClusterServer.GetCluster(ctx, &openapidedicatedv1beta1.GetClusterRequest{
			ClusterId: clusterID,
		})
		require.NoError(t, err)
		assert.Equal(t, openapicommonv1beta1.ClusterState_ACTIVE, cluster.State)

		err = openAPIOrchestrator.Commit(ctx, clusterID, requests, nil)
		require.NoError(t, err)

		for _, item := range requests {
			validateOnly, err := getValidateOnlyField(item.Message)
			require.NoError(t, err)
			assert.False(t, validateOnly)
		}

		instances, err := workflowimpl.GetWorkflow().ListAllRunningDagInstances()
		require.NoError(t, err)

		ins, ok := lo.Find(instances, func(item *entity.DagInstance) bool {
			ok := strings.HasPrefix(item.DagID, workflowutil.DynamicDagPrefix)
			if !ok {
				return false
			}

			_, ok = lo.Find(item.Tags, func(item entity.DagInstanceTag) bool {
				return item.Key == workflowutil.BizIdKey && item.Value == clusterID
			})
			return ok
		})
		assert.True(t, ok)

		// wait instance done
		time.Sleep(3 * time.Second)

		var dagInstance entity.DagInstance
		err = db.NewProvider(nil).
			ReadDB(ctx).
			Model(&entity.DagInstance{}).
			Where("id = ?", ins.ID).
			Take(&dagInstance).
			Error
		require.NoError(t, err)
		assert.True(
			t,
			dagInstance.Status == entity.DagInstanceStatusSuccess ||
				dagInstance.Status == entity.DagInstanceStatusRunning,
		)

		var tasks []entity.TaskInstance
		err = db.NewProvider(nil).
			ReadDB(ctx).
			Model(&entity.TaskInstance{}).
			Where("dag_ins_id = ?", ins.ID).
			Take(&tasks).
			Error
		require.NoError(t, err)

		_, ok = lo.Find(tasks, func(item entity.TaskInstance) bool {
			return item.Status == entity.TaskInstanceStatusSuccess
		})
		assert.True(t, ok)
	})
}

func getValidateOnlyField(req proto.Message) (bool, error) {
	rv := reflect.ValueOf(req)
	if rv.Kind() == reflect.Ptr {
		rv = rv.Elem()
	}

	if rv.Kind() != reflect.Struct {
		return false, errors.New("expect struct type")
	}

	field := rv.FieldByName("ValidateOnly")
	if !field.IsValid() {
		return false, errors.New("missing ValidateOnly field")
	}

	if field.Kind() != reflect.Bool {
		return false, errors.New("expect ValidateOnly as boolean type")
	}

	return field.Bool(), nil
}