package server

import (
	"context"
	"fmt"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

// CreateTiKVGroup creates a new TiKV group in the specified shared pool.
// It performs the following:
// 1. Validates the request parameters (tikv_group_id, node_quota, storages, labels)
// 2. Checks shared pool state and component uniqueness
// 3. Creates the component with CREATING phase
// 4. Submits the PlanCreateComponent DAG to create certificates, VAC, and apply the component
func (s *serverImpl) CreateTiKVGroup(ctx context.Context, sharedPoolId, tikvGroupId string, nodeQuota *entity.NodeQuota, storages []*entity.Storage, labels map[string]string, exclusive, force bool) (*entity.Component, error) {
	// Validate required parameters
	if sharedPoolId == "" {
		return nil, status.Errorf(codes.InvalidArgument, "shared_pool_id is required")
	}

	if tikvGroupId == "" {
		return nil, status.Errorf(codes.InvalidArgument, "tikv_group_id is required")
	}

	// Validate tikv_group_id pattern
	if err := validateTikvGroupId(tikvGroupId); err != nil {
		return nil, err
	}

	// Validate node_quota
	if err := validateTiKVGroupNodeQuota(nodeQuota); err != nil {
		return nil, err
	}

	// Validate storages
	if err := validateTiKVGroupStorages(storages, s.cloudProvider); err != nil {
		return nil, err
	}

	// Validate labels
	if err := validateTiKVGroupLabels(labels); err != nil {
		return nil, err
	}

	var component *entity.Component
	var dagExecId string
	var dagSubmitted bool

	if err := s.transactioner.Transaction(ctx, func(ctx context.Context) error {
		// Lock the shared pool
		unlockFunc, err := s.lock(ctx, sharedPoolId)
		if err != nil {
			tilog.Error(ctx, "failed to lock sharedPool", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
			return err
		}
		defer func() {
			if err := unlockFunc(); err != nil {
				tilog.Error(ctx, "failed to unlock sharedPool", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
			}
		}()

		// Load shared pool
		sharedPool, err := s.sharedPoolRepo.Load(ctx, sharedPoolId, false)
		if err != nil {
			if status.Code(err) != codes.NotFound {
				return status.Errorf(codes.Internal, "failed to load sharedPool: %s", err)
			}
			return status.Errorf(codes.NotFound, "sharedPool %s not found", sharedPoolId)
		}
		if err := guardSharedPoolMutation(sharedPool, sharedPoolMutationBlockedWhilePaused); err != nil {
			return err
		}

		// Merge shared pool status from dag task
		err = s.mergeSharedPoolStatusFromDagTask(ctx, sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to merge sharedPool status from dag task: %s", err)
		}
		if err := s.guardTiKVRemoteWorkersTopologyStable(ctx, sharedPool); err != nil {
			return err
		}

		// Check shared pool can accept new components
		if err := checkSharedPoolCanAcceptNewComponent(sharedPool); err != nil && (!force || sharedPool.GetPhase() != apiv1.SharedPool_UNAVAILABLE) {
			return err
		}

		// Check tikv_group_id uniqueness within the shared pool
		if err := checkTikvGroupIdUniqueness(sharedPool, tikvGroupId); err != nil {
			return err
		}

		// Get node spec map from dataplane
		nodeSpecMap, err := s.genNodeSpecMap(ctx, sharedPool)
		if err != nil {
			tilog.Error(ctx, "failed to get node spec map", zap.String("dataplaneId", sharedPool.GetDataplaneName()), zap.Error(err))
			return status.Errorf(codes.Internal, "failed to get node spec map: %s", err)
		}

		// Validate and fill node spec
		withLocalDisk := false
		for _, storage := range storages {
			if storage.GetName() == apiv1.Component_Storage_DATA && storage.GetType() == entity.LocalSSD {
				withLocalDisk = true
				break
			}
		}

		fullNodeSpec, err := getNodeSpec(ctx, nodeQuota.GetNodeSpec(), nodeSpecMap, withLocalDisk)
		if err != nil {
			return status.Errorf(codes.InvalidArgument, "node spec validation failed: %s", err)
		}

		// For local-ssd, validate node spec has local disk and update storage size
		if withLocalDisk {
			if !fullNodeSpec.GetWithLocalDisk() {
				return status.Errorf(codes.InvalidArgument, "node spec must have with_local_disk=true when using local-ssd storage")
			}
			// Post-validation size assignment for local-ssd:
			// ValidateStorageTypeForProvider requires SizeGi=0 for local-ssd (users can't specify size),
			// but we need to populate the actual size here for downstream processing.
			// See ValidateStorageParams documentation for the two-phase validation pattern.
			for _, storage := range storages {
				if storage.GetName() == apiv1.Component_Storage_DATA {
					storage.SizeGi = fullNodeSpec.GetLocalDiskSizeGi()
				}
			}
		}

		// Fill default mount path for RAFTLOG if not specified
		for _, storage := range storages {
			if storage.GetName() == apiv1.Component_Storage_RAFTLOG && storage.GetMountPath() == "" {
				storage.MountPath = entity.DefaultRaftlogMountPath
			}
		}

		// Create the component
		component = &entity.Component{
			Id:            tikvGroupId,
			ComponentType: apiv1.ComponentType_TIKV,
			NodeQuota: &apiv1.NodeQuota{
				NodeSpec:  fullNodeSpec,
				NodeCount: nodeQuota.GetNodeCount(),
			},
			Storages: storages,
			Labels:   labels,
			Phase:    apiv1.Component_CREATING,
		}

		// Append component to shared pool
		sharedPool.ComponentGroups = append(sharedPool.ComponentGroups, component)
		sharedPool.Phase = apiv1.SharedPool_MODIFYING

		// Save shared pool
		err = s.sharedPoolRepo.Save(ctx, sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to save sharedPool: %s", err)
		}

		// Prepare DAG arguments
		sharedPoolStr, err := protojson.Marshal(sharedPool)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to marshal sharedPool: %s", err)
		}

		args := actions.ActionArgs{
			workflow.SharedPoolIdKey:   actions.ActionArgString(sharedPool.GetId()),
			workflow.SharedPoolKey:     actions.ActionArgString(string(sharedPoolStr)),
			workflow.ComponentIdsKey:   actions.ActionArgString(tikvGroupId),
			workflow.ComponentTypesKey: actions.ActionArgString(entity.GetComponentTypeName(component)),
			workflow.MeteringTypeKey:   actions.ActionArgString(workflow.MeteringTypeValueLogicMetaBySharedPool),
			utils.TraceIdKey:           utils.GetTraceId(ctx),
		}
		if exclusive {
			args[workflow.TiKVPlacementExclusiveKey] = actions.ActionArgBool(true)
		}

		// Generate DAG execution ID
		dagExecId = utils.GeneratedDagExecId(ctx,
			fmt.Sprintf("create-tikv-group-%s-%s-%s",
				sharedPool.GetId(),
				tikvGroupId,
				utils.FnvHash(string(sharedPoolStr)),
			))

		// Submit DAG
		err = s.dagExecutor.Submit(ctx, dag.PlanCreateComponent, args, dagExecId)
		if err != nil {
			return status.Errorf(codes.Internal, "failed to submit create-tikv-group dag: %s", err)
		}
		dagSubmitted = true

		tilog.Info(ctx, "submitted create tikv group dag",
			zap.String("sharedPoolId", sharedPoolId),
			zap.String("tikvGroupId", tikvGroupId),
			zap.String("dagExecId", dagExecId))

		return nil
	}); err != nil {
		if dagSubmitted {
			if abortErr := s.dagExecutor.Abort(ctx, dagExecId); abortErr != nil {
				tilog.Error(ctx, "failed to abort dag after transaction failure",
					zap.String("dagExecId", dagExecId),
					zap.Error(abortErr))
			}
		}
		return nil, err
	}

	return component, nil
}

// validateTikvGroupId validates the tikv_group_id format
func validateTikvGroupId(tikvGroupId string) error {
	if err := entity.ValidateLowercaseAlphanumericId("tikv_group_id", tikvGroupId); err != nil {
		return status.Errorf(codes.InvalidArgument, "%s", err.Error())
	}
	return nil
}

// validateTiKVGroupNodeQuota validates node quota for TiKV group
func validateTiKVGroupNodeQuota(nodeQuota *entity.NodeQuota) error {
	if nodeQuota == nil {
		return status.Errorf(codes.InvalidArgument, "node_quota is required")
	}
	if nodeQuota.GetNodeSpec() == nil {
		return status.Errorf(codes.InvalidArgument, "node_quota.node_spec is required")
	}
	if nodeQuota.GetNodeSpec().GetCpu() == 0 {
		return status.Errorf(codes.InvalidArgument, "node_quota.node_spec.cpu is required")
	}
	if nodeQuota.GetNodeSpec().GetMemoryGi() == 0 {
		return status.Errorf(codes.InvalidArgument, "node_quota.node_spec.memory_gi is required")
	}
	if nodeQuota.GetNodeCount() == 0 {
		return status.Errorf(codes.InvalidArgument, "node_quota.node_count is required")
	}
	// TiKV node count must be multiple of 3
	if nodeQuota.GetNodeCount()%3 != 0 {
		return status.Errorf(codes.InvalidArgument, "node_quota.node_count must be multiple of 3, got %d", nodeQuota.GetNodeCount())
	}
	return nil
}

// validateTiKVGroupStorages validates storage configuration for TiKV group
func validateTiKVGroupStorages(storages []*entity.Storage, provider string) error {
	if len(storages) == 0 {
		return status.Errorf(codes.InvalidArgument, "storages are required")
	}

	hasData := false
	hasRaftlog := false

	for _, storage := range storages {
		switch storage.GetName() {
		case apiv1.Component_Storage_DATA:
			hasData = true
		case apiv1.Component_Storage_RAFTLOG:
			hasRaftlog = true
			// RAFTLOG cannot use local-ssd because:
			// 1. CreateVolumeAttributesClassAction skips LocalSSD entirely
			// 2. ApplyComponentGroupAction always sets RaftLogVolumeAttributesClass
			// This would leave a non-existent VAC referenced in the template
			if storage.GetType() == entity.LocalSSD {
				return status.Errorf(codes.InvalidArgument, "RAFTLOG storage cannot use local-ssd type")
			}
		default:
			return status.Errorf(codes.InvalidArgument, "invalid storage name '%s': only DATA and RAFTLOG are allowed", storage.GetName().String())
		}

		// Validate storage type based on cloud provider using unified validation
		if err := entity.ValidateStorageTypeForProvider(storage, provider, storage.GetName().String()); err != nil {
			return status.Errorf(codes.InvalidArgument, "%s", err.Error())
		}
	}

	if !hasData {
		return status.Errorf(codes.InvalidArgument, "DATA storage is required")
	}
	if !hasRaftlog {
		return status.Errorf(codes.InvalidArgument, "RAFTLOG storage is required")
	}

	return nil
}

// validateTiKVGroupLabels validates labels for TiKV group
func validateTiKVGroupLabels(labels map[string]string) error {
	if len(labels) == 0 {
		return nil
	}

	// Validate label count
	if len(labels) > MaxLabelsCount {
		return status.Errorf(codes.InvalidArgument, "too many labels: %d exceeds maximum of %d", len(labels), MaxLabelsCount)
	}

	for key, value := range labels {
		// Check for reserved label keys
		if isReservedLabelKey(key) {
			return status.Errorf(codes.InvalidArgument, "label key '%s' is reserved and cannot be set externally", key)
		}
		// Validate key format and length
		if err := validateLabelKey(key); err != nil {
			return status.Errorf(codes.InvalidArgument, "invalid label key: %s", err.Error())
		}
		// Validate value length
		if err := validateLabelValue(key, value); err != nil {
			return status.Errorf(codes.InvalidArgument, "invalid label value: %s", err.Error())
		}
	}

	return nil
}

// checkSharedPoolCanAcceptNewComponent checks if the shared pool can accept new components
func checkSharedPoolCanAcceptNewComponent(sharedPool *entity.SharedPool) error {
	switch sharedPool.GetPhase() {
	case apiv1.SharedPool_CREATING:
		return status.Errorf(codes.FailedPrecondition, "shared pool is creating, please try again later")
	case apiv1.SharedPool_PAUSING:
		return status.Errorf(codes.FailedPrecondition, "shared pool is pausing, cannot create new components")
	case apiv1.SharedPool_UNAVAILABLE:
		return status.Errorf(codes.FailedPrecondition, "shared pool is unavailable, cannot create new components")
	case apiv1.SharedPool_DELETING:
		return status.Errorf(codes.FailedPrecondition, "shared pool is being deleted, cannot create new components")
	}
	return nil
}

// checkTikvGroupIdUniqueness checks if the tikv_group_id is unique within the shared pool
func checkTikvGroupIdUniqueness(sharedPool *entity.SharedPool, tikvGroupId string) error {
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetId() == tikvGroupId && comp.GetComponentType() == apiv1.ComponentType_TIKV {
			return status.Errorf(codes.AlreadyExists, "tikv group with id '%s' already exists in shared pool", tikvGroupId)
		}
	}
	return nil
}