package action

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/google/uuid"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr"
	"github.com/tidbcloud/cluster-service-ng/pkg/nodegroup"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"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"
)

const (
	// CreateNodeInstanceGroupActionName is the action name for creating node instance group
	CreateNodeInstanceGroupActionName = "create-node-instance-group-action"

	// DefaultLockLeaseDuration is the default lease duration for distributed lock
	DefaultLockLeaseDuration = 5 * time.Minute
)

// CreateNodeInstanceGroupAction adds a new node instance group to a SharedPool.
type CreateNodeInstanceGroupAction struct {
	nodeGroupStateRepo     repo.DataplaneNodeInstanceGroupStateRepo
	nodeGroupLockRepo      repo.DataplaneNodeInstanceGroupLockRepo
	dataplaneServiceClient dplanemgrv1.DataPlaneServiceClient
}

// NewCreateNodeInstanceGroupAction creates a new CreateNodeInstanceGroupAction.
func NewCreateNodeInstanceGroupAction(
	nodeGroupStateRepo repo.DataplaneNodeInstanceGroupStateRepo,
	nodeGroupLockRepo repo.DataplaneNodeInstanceGroupLockRepo,
	dataplaneClient dataplanemgr.Client,
) actions.Action {
	return &CreateNodeInstanceGroupAction{
		nodeGroupStateRepo:     nodeGroupStateRepo,
		nodeGroupLockRepo:      nodeGroupLockRepo,
		dataplaneServiceClient: dataplaneClient.GetDataPlaneServiceClient(),
	}
}

// Name returns the action name.
func (a *CreateNodeInstanceGroupAction) Name() string {
	return CreateNodeInstanceGroupActionName
}

// Invoke executes the create node instance group action.
func (a *CreateNodeInstanceGroupAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, "create node instance group action invoked")

	// Parse arguments
	dataplaneName, err := args.GetString(workflow.DataplaneIdKey)
	if err != nil {
		return fmt.Errorf("failed to get dataplane_id: %w", err)
	}

	// Parse NodeInstanceGroup from serialized data
	groupData, err := args.GetString(workflow.NodeInstanceGroupKey)
	if err != nil {
		return fmt.Errorf("failed to get node_instance_group: %w", err)
	}
	var group entity.NodeInstanceGroup
	if err := protojson.Unmarshal([]byte(groupData), &group); err != nil {
		return status.Errorf(codes.InvalidArgument, "failed to unmarshal node instance group: %v", err)
	}

	nodeSpec := entity.GetComponentNodeSpec(group.GetNodeSpec())
	primaryInstance := group.GetPrimaryInstanceType()
	primaryPriority := group.GetPriority()

	// Validate primary priority (must be >= 1000 if set)
	if err := nodegroup.ValidatePriority(primaryPriority, true); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid primary priority: %v", err)
	}

	// Build primary spec
	primary := nodegroup.InstanceTypeSpec{
		Spec:         nodeSpec,
		InstanceType: primaryInstance,
		Priority:     primaryPriority,
	}
	if group.GetNodeSpec().GetAvailableCpu() > 0 {
		primary.AvailableCPU = fmt.Sprintf("%d", group.GetNodeSpec().GetAvailableCpu())
	}
	if group.GetNodeSpec().GetAvailableMemoryMi() > 0 {
		primary.AvailableMemoryMi = fmt.Sprintf("%d", group.GetNodeSpec().GetAvailableMemoryMi())
	}

	if dataplaneName == "" {
		return status.Errorf(codes.InvalidArgument, "dataplane_id is required")
	}
	if err := nodegroup.ValidateSpec(primary.Spec); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid node_spec: %v", err)
	}
	if err := nodegroup.ValidateCloudInstanceType(primary.InstanceType); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid primary_instance: %v", err)
	}
	if primary.AvailableCPU != "" {
		if err := nodegroup.ValidateAvailableCPU(primary.AvailableCPU); err != nil {
			return status.Errorf(codes.InvalidArgument, "invalid available_cpu: %v", err)
		}
	}
	// Note: AvailableMemory contains MiB values (from AvailableMemoryMi proto field).
	// We skip GiB-range validation here since MiB values (e.g., 15360 for ~15Gi) would
	// fail the 1-2048 GiB check. The proto uint32 type provides sufficient validation.
	withLocalDisk := group.GetNodeSpec().GetWithLocalDisk()
	localDiskSizeGi := group.GetNodeSpec().GetLocalDiskSizeGi()

	// placement_group_enabled is the durable per-NodePool gate sent on the
	// wire to DPM, which now honours it (with a topology sanity check). The
	// resolution chain is:
	//
	//  1. If a non-builtin sibling NodePool exists in the dataplane, the
	//     caller inherits Spec.PlacementGroupEnabled from it. This is the
	//     authoritative source — a legacy single-AZ, PG-disabled primary
	//     keeps its fallback on the legacy per-zone path even though the
	//     dataplane topology is itself eligible for partition PGs.
	//  2. Otherwise (first NodePool in the dataplane), fall back to the
	//     DAG arg (= SharedPool.Provider.PlacementGroupEnabled, set at
	//     CreateSharedPool time from "AWS + single-AZ?"). For new
	//     SharedPools this matches the topology derivation; for legacy
	//     SharedPools this is the legacy false record that must persist.
	//
	// Built-in NodePools are excluded from sibling inheritance via the
	// workload-pool label discriminator (nodegroup.LabelNodeSpec), since
	// they are not part of the workload's PG scope.
	placementGroupEnabled, err := a.resolvePlacementGroupEnabled(ctx, dataplaneName, args)
	if err != nil {
		// Forward the underlying gRPC code (InvalidArgument for malformed
		// DAG args) rather than wrapping as Internal, so callers see the
		// right category.
		return err
	}

	// Generate unique owner ID with UUID to prevent concurrent invocations from sharing the same lock.
	// This ensures true mutual exclusion even when the same spec is processed concurrently.
	ownerID := fmt.Sprintf("%s-%s-%s", CreateNodeInstanceGroupActionName, nodeSpec, uuid.New().String()[:8])
	lock, err := a.nodeGroupLockRepo.TryAcquire(ctx, dataplaneName, nodeSpec, primaryInstance, ownerID, DefaultLockLeaseDuration)
	if err != nil {
		tilog.Error(ctx, "failed to acquire lock",
			zap.String("dataplaneName", dataplaneName),
			zap.String("nodeSpec", nodeSpec),
			zap.String("ownerID", ownerID),
			zap.Error(err))
		return status.Errorf(codes.Unavailable, "failed to acquire lock: %v", err)
	}
	if lock == nil {
		tilog.Warn(ctx, "lock is held by another owner",
			zap.String("dataplaneName", dataplaneName),
			zap.String("nodeSpec", nodeSpec),
			zap.String("ownerID", ownerID))
		return status.Errorf(codes.Aborted, "lock is held by another owner, retry later")
	}

	defer func() {
		if releaseErr := a.nodeGroupLockRepo.Release(ctx, dataplaneName, nodeSpec, primaryInstance, ownerID); releaseErr != nil {
			tilog.Warn(ctx, "failed to release lock",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.String("ownerID", ownerID),
				zap.Error(releaseErr))
		}
	}()

	tilog.Info(ctx, "acquired distributed lock",
		zap.String("dataplaneName", dataplaneName),
		zap.String("nodeSpec", nodeSpec),
		zap.String("ownerID", ownerID),
		zap.Int64("fencingToken", lock.FencingToken))

	return a.createNodeInstanceGroup(ctx, dataplaneName, &primary, withLocalDisk, localDiskSizeGi, placementGroupEnabled, group.GetNodeSpec())
}

func (a *CreateNodeInstanceGroupAction) createNodeInstanceGroup(
	ctx context.Context,
	dataplaneName string,
	primary *nodegroup.InstanceTypeSpec,
	withLocalDisk bool,
	localDiskSizeGi uint32,
	placementGroupEnabled bool,
	nodeSpecProto *entity.NodeSpec,
) error {
	// 2. Use resource prefix "primary" for primary nodepools.
	// DPM builds the full NodePool CR name by appending instance/spec/dataplane/hash/k8s-version.
	resourcePrefix := "primary"

	// 3. Build labels (for reference - will need K8s API to apply)
	labels := nodegroup.BuildPrimaryLabels(primary)

	tilog.Info(ctx, "creating node instance group nodepool",
		zap.String("dataplaneName", dataplaneName),
		zap.String("spec", primary.Spec),
		zap.String("instanceType", primary.InstanceType),
		zap.Any("labels", labels))

	cpuCore, memoryGi, err := nodegroup.ParseSpec(primary.Spec)
	if err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid node_spec: %v", err)
	}

	availableCPUCore := uint32(0)
	if primary.AvailableCPU != "" {
		val, err := strconv.Atoi(primary.AvailableCPU)
		if err != nil {
			return status.Errorf(codes.InvalidArgument, "invalid available_cpu: %v", err)
		}
		availableCPUCore = uint32(val)
	}
	availableMemoryMi := uint32(0)
	if primary.AvailableMemoryMi != "" {
		val, err := strconv.Atoi(primary.AvailableMemoryMi)
		if err != nil {
			return status.Errorf(codes.InvalidArgument, "invalid available_memory_mi: %v", err)
		}
		availableMemoryMi = uint32(val)
	}

	availableCPUMilli := availableCPUCore * 1000
	if nodeSpecProto != nil && nodeSpecProto.GetAvailableCpuMilli() > 0 {
		availableCPUMilli = nodeSpecProto.GetAvailableCpuMilli()
	}
	maxCPUMilli := availableCPUMilli
	if nodeSpecProto != nil && nodeSpecProto.GetMaxCpuMilli() > 0 {
		maxCPUMilli = nodeSpecProto.GetMaxCpuMilli()
	}
	maxMemoryMi := availableMemoryMi
	if nodeSpecProto != nil && nodeSpecProto.GetMaxMemoryMi() > 0 {
		maxMemoryMi = nodeSpecProto.GetMaxMemoryMi()
	}

	// 4. Check dataplane status before creating nodepool - skip if deleting or dataplane not found
	dp, err := a.dataplaneServiceClient.GetDataPlane(ctx, &dplanemgrv1.GetDataPlaneRequest{DataPlaneId: dataplaneName})
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Warn(ctx, "dataplane not found when creating nodepool, skipping",
				zap.String("dataplaneName", dataplaneName),
				zap.String("spec", primary.Spec),
				zap.String("instanceType", primary.InstanceType),
				zap.Error(err))
			return nil
		}
		return status.Errorf(codes.Internal, "failed to get dataplane status: %v", err)
	}
	if dp.GetStatus() == dplanemgrv1.DataPlane_Deleting {
		tilog.Info(ctx, "skipping nodepool creation for deleting dataplane",
			zap.String("dataplaneName", dataplaneName))
		return nil
	}

	// 5. Call DPM to create nodepool
	req := &dplanemgrv1.CreateNodePoolRequest{
		DataPlaneId:    dataplaneName,
		ResourcePrefix: resourcePrefix,
		NodePool: &dplanemgrv1.NodePool{
			InstanceType:          primary.InstanceType,
			NodeType:              dplanemgrv1.NodeType_Dedicated,
			CpuCore:               cpuCore,
			MemoryGi:              memoryGi,
			AvailableCpuMilliCore: availableCPUMilli,
			MaxCpuMilli:           maxCPUMilli,
			AvailableMemoryMi:     availableMemoryMi,
			MaxMemoryMi:           maxMemoryMi,
			WithLocalDisk:         withLocalDisk,
			LocalDiskSizeGi:       localDiskSizeGi,
			// placement_group_enabled is the durable per-NodePool gate
			// resolved by resolvePlacementGroupEnabled above (sibling
			// inheritance → DAG arg fallback). The DPM service honours the
			// value with a topology sanity check, so a sibling NodePool's
			// legacy false record propagates onto a new sibling in the
			// same dataplane instead of being overwritten by the
			// topology-derived value.
			PlacementGroupEnabled: placementGroupEnabled,
			Labels:                labels,
		},
	}

	_, err = a.dataplaneServiceClient.CreateNodePool(ctx, req)
	if err != nil {
		if isAlreadyExistsError(err) {
			tilog.Info(ctx, "nodepool already exists, skipping creation",
				zap.String("dataplaneName", dataplaneName),
				zap.String("spec", primary.Spec),
				zap.String("instanceType", primary.InstanceType))
		} else {
			tilog.Error(ctx, "failed to create nodepool",
				zap.String("dataplaneName", dataplaneName),
				zap.String("instanceType", primary.InstanceType),
				zap.Error(err))
			return status.Errorf(codes.Internal, "failed to create nodepool: %v", err)
		}
	}

	// 6. Initialize NodeInstanceGroupState for shortage detection
	if _, err := a.nodeGroupStateRepo.LoadOrCreate(ctx, dataplaneName, primary.Spec, primary.InstanceType); err != nil {
		tilog.Warn(ctx, "failed to initialize node instance group state",
			zap.String("dataplaneName", dataplaneName),
			zap.String("primarySpec", primary.Spec),
			zap.Error(err))
	}

	tilog.Info(ctx, "successfully created node instance group",
		zap.String("dataplaneName", dataplaneName),
		zap.String("spec", primary.Spec),
		zap.String("instanceType", primary.InstanceType))

	return nil
}

// resolvePlacementGroupEnabled determines the value to send on the
// CreateNodePool wire for placement_group_enabled. The chain is:
//
//  1. Sibling inheritance: list existing NodePools in the dataplane
//     (skipping built-in admin/o11y pools, which carry no Spec.Labels)
//     and inherit Spec.PlacementGroupEnabled from the first workload
//     sibling found. Workload pools are identified by their spec label
//     nodegroup.LabelNodeSpec, which nodegroup.BuildPrimaryLabels sets on
//     every workload pool but the dataplane controller never sets on
//     built-ins. This is the authoritative source for any dataplane with
//     at least one existing workload pool, because legacy single-AZ,
//     PG-disabled pools must keep their durable false record even
//     though the dataplane topology is itself eligible for partition PGs.
//  2. DAG-arg fallback: if no workload sibling exists (first NodePool in
//     the dataplane), use the DAG arg (= SharedPool.Provider.Placement
//     GroupEnabled, set at CreateSharedPool time from "AWS + single-AZ?").
//     For new SharedPools this matches the topology derivation; for
//     legacy SharedPools it is the legacy false record.
//  3. Default false: if the DAG arg is missing too (extreme edge case
//     where the SharedPool was never persisted), fall back to false. The
//     DPM topology sanity check ensures a wrong true can never reach the
//     spec; false is always safe.
//
// ListNodePools failures are logged and treated as "no sibling": the DAG
// arg fallback still produces a valid durable record (false-or-true
// based on the SharedPool gate), and a transient RPC error must not block
// NodePool creation.
func (a *CreateNodeInstanceGroupAction) resolvePlacementGroupEnabled(ctx context.Context, dataplaneName string, args actions.ActionArgs) (bool, error) {
	placementGroupEnabled := false
	if _, ok := args[workflow.PlacementGroupEnabledKey]; ok {
		var err error
		placementGroupEnabled, err = args.GetBool(workflow.PlacementGroupEnabledKey)
		if err != nil {
			// Malformed arg: propagate as InvalidArgument so the DAG surfaces
			// a clear error to the caller, not a generic Internal. This
			// matches the contract that the DAG arg channel is type-checked.
			return false, status.Errorf(codes.InvalidArgument, "invalid placement_group_enabled: %v", err)
		}
	}

	resp, err := a.dataplaneServiceClient.ListNodePools(ctx, &dplanemgrv1.ListNodePoolsRequest{
		DataPlaneId: dataplaneName,
	})
	if err != nil {
		// Transient RPC error: log and fall back to the DAG arg. The DAG
		// arg is the SharedPool gate, which is itself durable; a flaky DPM
		// must not block NodePool creation. The DPM topology sanity check
		// on the receiving side ensures a wrong true can never reach the
		// spec; false is always safe.
		tilog.Warn(ctx, "failed to list existing nodepools for sibling inheritance; falling back to SharedPool gate",
			zap.String("dataplaneName", dataplaneName),
			zap.Error(err))
		return placementGroupEnabled, nil
	}
	for _, pool := range resp.GetNodePools() {
		if pool == nil {
			continue
		}
		// Built-in NodePools (admin / o11y) are managed by the dataplane
		// controller, have Spec.PlacementGroupEnabled == false by default,
		// and carry no Spec.Labels. Workload pools carry spec labels built
		// by nodegroup.BuildPrimaryLabels, which include LabelNodeSpec.
		// Using LabelNodeSpec as the workload discriminator avoids
		// depending on K8s annotations (not exposed on the gRPC NodePool
		// proto) and matches the same name the dataplane controller reads
		// when classifying pools.
		if pool.GetLabels()[nodegroup.LabelNodeSpec] == "" {
			continue
		}
		tilog.Info(ctx, "inheriting placement_group_enabled from sibling nodepool",
			zap.String("dataplaneName", dataplaneName),
			zap.String("siblingPoolName", pool.GetName()),
			zap.Bool("inherited", pool.GetPlacementGroupEnabled()))
		return pool.GetPlacementGroupEnabled(), nil
	}
	return placementGroupEnabled, nil
}

// isAlreadyExistsError checks if the error indicates the resource already exists.
// Only relies on gRPC status code for robust detection.
func isAlreadyExistsError(err error) bool {
	if err == nil {
		return false
	}

	// Only rely on gRPC status code - do not use error message string matching
	// as it's fragile and can break when DPM changes error messages.
	if st, ok := status.FromError(err); ok {
		return st.Code() == codes.AlreadyExists
	}

	return false
}

// nodePoolTaintsToNodeGroup converts DPM NodePoolTaint slice to nodegroup.Taint slice.
func nodePoolTaintsToNodeGroup(taints []*dplanemgrv1.NodePoolTaint) []nodegroup.Taint {
	result := make([]nodegroup.Taint, 0, len(taints))
	for _, t := range taints {
		if t == nil {
			continue
		}
		result = append(result, nodegroup.Taint{
			Key:    t.GetKey(),
			Value:  t.GetValue(),
			Effect: t.GetEffect(),
		})
	}
	return result
}

// nodeGroupTaintsToNodePool converts nodegroup.Taint slice to DPM NodePoolTaint slice.
func nodeGroupTaintsToNodePool(taints []nodegroup.Taint) []*dplanemgrv1.NodePoolTaint {
	result := make([]*dplanemgrv1.NodePoolTaint, 0, len(taints))
	for _, t := range taints {
		result = append(result, &dplanemgrv1.NodePoolTaint{
			Key:    t.Key,
			Value:  t.Value,
			Effect: t.Effect,
		})
	}
	return result
}