package orchestrator

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"strconv"

	"github.com/samber/lo"
	"github.com/tidbcloud/fastflow"
	"github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/fastflow/pkg/entity/run"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	bffclusterv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/bff/cluster_management/v1beta1"
	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"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	DagConsoleAPIScale = "console_api_scale"
)

type APIOrchestrator struct {
	workflow                      workflow.Workflow
	customServer                  tidbmgmtservicev1.ClusterCustomStatusServiceServer
	openAPIDedicatedClusterServer openapidedicatedv1beta1.ClusterServiceServer
}

type OpenAPIRequest struct {
	Message    proto.Message
	ActionName ActionName
}

func NewAPIOrchestrator(
	wflow workflow.Workflow,
	customServer tidbmgmtservicev1.ClusterCustomStatusServiceServer,
	openAPIDedicatedClusterServer openapidedicatedv1beta1.ClusterServiceServer,
	bizActions []*OpenAPIAction,
) *APIOrchestrator {
	actions := []run.Action{
		NewWaitClusterNormalAction(openAPIDedicatedClusterServer),
		NewDeleteClusterCustomStatus(customServer),
	}
	lo.ForEach(bizActions, func(item *OpenAPIAction, index int) {
		actions = append(actions, item)
	})
	fastflow.RegisterAction(actions)
	return &APIOrchestrator{
		workflow:                      wflow,
		customServer:                  customServer,
		openAPIDedicatedClusterServer: openAPIDedicatedClusterServer,
	}
}

// Caution: Each component is validated separately!
func (a *APIOrchestrator) ValidateOneByOne(ctx context.Context, requests []*OpenAPIRequest) error {
	for _, req := range requests {
		err := setValidateOnlyField(ctx, req.Message, true)
		if err != nil {
			tilog.Error(ctx, "set ValidateOnly field failed", zap.Error(err))
			return errors.New("request failed")
		}

		action, ok := fastflow.GetAction(string(req.ActionName))
		if !ok {
			tilog.Error(ctx, "not found action", zap.Any("action", req.ActionName))
			return errors.New("request failed")
		}

		openAPIAction, ok := action.(*OpenAPIAction)
		if !ok {
			tilog.Error(ctx, "expect OpenAPIAction type", zap.Any("action", req.ActionName))
			return errors.New("request failed")
		}

		data, err := protojson.Marshal(req.Message)
		if err != nil {
			tilog.Error(ctx, "marshal request failed", zap.Error(err))
			return err
		}
		if err := openAPIAction.action(ctx, string(data)); err != nil {
			tilog.Error(ctx, "request failed", zap.Error(err))
			return err
		}
	}

	return nil
}

func (a *APIOrchestrator) Commit(
	ctx context.Context,
	clusterId string,
	requests []*OpenAPIRequest,
	sourceReq *bffclusterv1beta1.AdvancedModifyDedicatedClusterRequest,
) error {
	cluster, err := a.openAPIDedicatedClusterServer.GetCluster(ctx, &openapidedicatedv1beta1.GetClusterRequest{
		ClusterId: clusterId,
	})
	if err != nil {
		return err
	}
	if cluster.State != openapicommonv1beta1.ClusterState_ACTIVE {
		return errors.New("cluster is not active state")
	}

	mutex := &tidbmgmtservicev1.ClusterCustomStatus_CustomMutex{
		ClusterAction: []tidbmgmtservicev1.ClusterAction{
			tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_SCALE_CLUSTER,
			tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
			tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_MODIFY_CLUSTER,
		},
	}

	id, err := strconv.ParseUint(clusterId, 10, 64)
	if err != nil {
		return err
	}
	status, err := a.customServer.SetClusterCustomStatus(ctx, &tidbmgmtservicev1.SetClusterCustomStatusRequest{
		CustomStatus: &tidbmgmtservicev1.ClusterCustomStatus{
			ClusterId:   id,
			StatusName:  fmt.Sprintf("scale-cluster-%s-%s", clusterId, lo.RandomString(5, lo.AllCharset)), // random
			Mutex:       mutex,
			TimeoutSecs: nil,
			StatusType:  tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_MODIFYING,
		},
	})
	if err != nil {
		return err
	}

	tasks, err := transformTasks(ctx, requests)
	if err != nil {
		return err
	}

	metadataJson, err := metadata.MarshalRequestMetaFromCtx(ctx)
	if err != nil {
		return fmt.Errorf("failed to marshal request metadata: %w", err)
	}

	// get source request
	sourceReqJson, err := protojson.Marshal(sourceReq)
	if err != nil {
		return fmt.Errorf("failed to marshal source request: %w", err)
	}

	return a.workflow.RegisterDynamicDagThenRun(ctx, DagConsoleAPIScale, &workflow.CreateDynamicDagOption{
		OrgId:     cluster.Labels[common.OrgIDLabelKey],
		ProjectId: cluster.Labels[common.ProjectIDLabelKey],
		BizId:     clusterId,
		DagVars: map[string]string{
			DagKeyClusterId:             clusterId,
			DagKeyClusterCustomStatusId: fmt.Sprintf("%d", status.StatusId),
			workflow.KeyRequestMeta:     string(metadataJson),
			DagKeyRequest:               string(sourceReqJson),
			workflow.KeyTraceId:         workflow.GetTraceId(ctx),
		},
		DagTasks: tasks,
	})
}

// ListRunningDagInstancesByNamePrefix
func (a *APIOrchestrator) ListRunningDagInstancesByNamePrefix(
	ctx context.Context,
	dagNamePrefix, bizId string,
) ([]*entity.DagInstance, error) {
	return a.workflow.ListRunningDagInstancesByNamePrefix(ctx, dagNamePrefix, bizId)
}

func setValidateOnlyField(ctx context.Context, req proto.Message, validateOnly bool) error {
	rv := reflect.ValueOf(req)
	if rv.Kind() == reflect.Ptr {
		rv = rv.Elem()
	}

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

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

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

	field.SetBool(validateOnly)
	return nil
}

func transformTasks(ctx context.Context, requests []*OpenAPIRequest) ([]entity.Task, error) {
	tasks := make([]entity.Task, 0)
	for i, req := range requests {
		err := setValidateOnlyField(ctx, req.Message, false)
		if err != nil {
			tilog.Error(ctx, "set ValidateOnly field failed", zap.Error(err))
			return nil, errors.New("request failed")
		}
		if string(req.ActionName) == "" {
			return nil, errors.New("miss action name")
		}

		task, err := openAPITask(
			fmt.Sprintf("%s-%d", req.ActionName, i),
			req.ActionName,
			req.Message,
			lo.IfF(len(tasks) > 0, func() string {
				return tasks[len(tasks)-1].ID
			}).Else(""),
		)
		if err != nil {
			return nil, err
		}

		tasks = append(tasks, task, waitClusterActiveTask(fmt.Sprintf("%s-%d", actionWaitClusterActive, i), task.ID))
	}

	return append(tasks, deleteCustomStatusTask(tasks[len(tasks)-1].ID)), nil
}

func openAPITask(id string, actionName ActionName, message proto.Message, parentTaskId string) (entity.Task, error) {
	data, err := protojson.Marshal(message)
	if err != nil {
		return entity.Task{}, err
	}
	task := entity.Task{
		ID:         id,
		Name:       id,
		ActionName: string(actionName),
		Params: map[string]any{
			TaskKeyRequestBody: string(data),
		},
	}
	if parentTaskId != "" {
		task.DependOn = []string{parentTaskId}
	}

	return task, nil
}

func waitClusterActiveTask(id, parentTaskId string) entity.Task {
	return entity.Task{
		ID:         id,
		Name:       id,
		ActionName: actionWaitClusterActive,
		DependOn:   []string{parentTaskId},
	}
}

func deleteCustomStatusTask(parentTaskId string) entity.Task {
	return entity.Task{
		ID:         actionDeleteCustomStatus,
		Name:       actionDeleteCustomStatus,
		ActionName: actionDeleteCustomStatus,
		DependOn:   []string{parentTaskId},
	}
}