package action

import (
	"context"
	"fmt"

	"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 (
	// DeprecateNodeInstanceGroupActionName is the action name for deprecating node instance group
	DeprecateNodeInstanceGroupActionName = "deprecate-node-instance-group-action"
)

// DeprecateNodeInstanceGroupAction deprecates a node instance group by adding NoSchedule taint.
// This applies to both primary and fallback nodepools.
type DeprecateNodeInstanceGroupAction struct {
	nodeGroupStateRepo     repo.DataplaneNodeInstanceGroupStateRepo
	nodeGroupLockRepo      repo.DataplaneNodeInstanceGroupLockRepo
	dataplaneServiceClient dplanemgrv1.DataPlaneServiceClient
}

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

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

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

	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()

	if dataplaneName == "" {
		return status.Errorf(codes.InvalidArgument, "dataplane_id is required")
	}
	if err := nodegroup.ValidateSpec(nodeSpec); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid node_spec: %v", err)
	}
	if primaryInstance != "" {
		if err := nodegroup.ValidateCloudInstanceType(primaryInstance); err != nil {
			return status.Errorf(codes.InvalidArgument, "invalid primary_instance: %v", err)
		}
	}

	ownerID := fmt.Sprintf("%s-%s-%s", DeprecateNodeInstanceGroupActionName, 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.deprecateNodeInstanceGroup(ctx, dataplaneName, nodeSpec, primaryInstance, group.GetFallbackInstances())
}

func (a *DeprecateNodeInstanceGroupAction) deprecateNodeInstanceGroup(
	ctx context.Context,
	dataplaneName, nodeSpec, primaryInstance string,
	fallbackInstances []*entity.NodeInstanceGroupFallbackInstance,
) error {
	tilog.Info(ctx, "deprecating node instance group",
		zap.String("dataplaneName", dataplaneName),
		zap.String("nodeSpec", nodeSpec),
		zap.String("primaryInstance", primaryInstance))

	// 1. Check state first for idempotency (avoid redundant taint operations on re-invocation)
	state, err := a.nodeGroupStateRepo.LoadOrCreate(ctx, dataplaneName, nodeSpec, primaryInstance)
	if err != nil {
		return status.Errorf(codes.Internal, "failed to load or create state: %v", err)
	}

	// Idempotency: if already deprecated, return success
	if state.State == entity.NodeInstanceGroupStateDeprecated {
		tilog.Info(ctx, "node instance group already deprecated, treating as success (idempotent)",
			zap.String("stateId", state.ID),
			zap.String("dataplaneName", dataplaneName),
			zap.String("nodeSpec", nodeSpec))
		return nil
	}

	if state.State != entity.NodeInstanceGroupStateNormal {
		tilog.Warn(ctx, "cannot deprecate: state is not normal",
			zap.String("stateId", state.ID),
			zap.String("currentState", state.State))
		return status.Errorf(codes.FailedPrecondition, "cannot deprecate: current state is %s, expected normal", state.State)
	}

	// 2. Verify the nodepool exists by listing nodepools
	nodePoolsResp, err := a.dataplaneServiceClient.ListNodePools(ctx, &dplanemgrv1.ListNodePoolsRequest{
		DataPlaneId: dataplaneName,
	})
	if err != nil {
		return status.Errorf(codes.Internal, "failed to list nodepools: %v", err)
	}

	// Build the set of instance types belonging to this group (primary + fallbacks).
	groupInstanceTypes := map[string]nodegroup.NodePoolRole{}
	if primaryInstance != "" {
		groupInstanceTypes[primaryInstance] = nodegroup.NodePoolRolePrimary
	}
	for _, fb := range fallbackInstances {
		if fb != nil && fb.GetInstanceType() != "" {
			if _, exists := groupInstanceTypes[fb.GetInstanceType()]; exists {
				continue
			}
			groupInstanceTypes[fb.GetInstanceType()] = nodegroup.NodePoolRoleFallback
		}
	}

	// Apply deprecation taint to nodepools that belong to this group.
	// Filter by node_spec label, instance type, and nodepool role
	// to avoid cross-group tainting when multiple NodeInstanceGroups share the same node_spec.
	// Also supports dataplane default pools matched by CPU/Memory.
	cpu, mem, err := nodegroup.ParseSpec(nodeSpec)
	if err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid node spec %s: %v", nodeSpec, err)
	}
	deprecatedTaint := nodegroup.DeprecatedTaint()
	for _, np := range nodePoolsResp.GetNodePools() {
		if np == nil {
			continue
		}
		labels := np.GetLabels()
		isMatch := false

		if nodegroup.IsManagedByInstanceTypeManager(labels) {
			// Case 1: InstanceTypeManager managed pools - match by node-spec label
			if labels != nil && labels[nodegroup.LabelNodeSpec] == nodeSpec {
				isMatch = true
			}
		} else {
			// Case 2: Dataplane default pools - match by CPU/Memory
			if np.GetCpuCore() == cpu && np.GetMemoryGi() == mem {
				isMatch = true
			}
		}

		if !isMatch {
			continue
		}
		if len(groupInstanceTypes) == 0 {
			if err := a.applyNodePoolTaint(ctx, dataplaneName, np, deprecatedTaint); err != nil {
				return err
			}
			continue
		}
		expectedRole, ok := groupInstanceTypes[np.GetInstanceType()]
		if !ok {
			continue
		}
		if !nodegroup.NodePoolMatchesRole(np, expectedRole) {
			tilog.Info(ctx, "skipping nodepool with mismatched role",
				zap.String("nodePoolName", np.GetName()),
				zap.String("expectedRole", string(expectedRole)),
				zap.String("instanceType", np.GetInstanceType()))
			continue
		}
		if err := a.applyNodePoolTaint(ctx, dataplaneName, np, deprecatedTaint); err != nil {
			return err
		}
	}

	// 3. Record deprecation in state
	if err := a.nodeGroupStateRepo.TransitionState(ctx, state.ID, entity.NodeInstanceGroupStateNormal, entity.NodeInstanceGroupStateDeprecated); err != nil {
		return status.Errorf(codes.Internal, "failed to transition to deprecated state: %v", err)
	}
	tilog.Info(ctx, "transitioned to deprecated state",
		zap.String("stateId", state.ID),
		zap.String("previousState", entity.NodeInstanceGroupStateNormal),
		zap.String("newState", entity.NodeInstanceGroupStateDeprecated))

	tilog.Info(ctx, "deprecation recorded",
		zap.String("dataplaneName", dataplaneName),
		zap.String("nodeSpec", nodeSpec),
		zap.String("stateId", state.ID),
		zap.String("taintKey", deprecatedTaint.Key),
		zap.String("taintValue", deprecatedTaint.Value))

	tilog.Info(ctx, "deprecate node instance group action completed",
		zap.String("dataplaneName", dataplaneName),
		zap.String("nodeSpec", nodeSpec))

	return nil
}

func (a *DeprecateNodeInstanceGroupAction) applyNodePoolTaint(
	ctx context.Context,
	dataplaneName string,
	nodePool *dplanemgrv1.NodePool,
	taint nodegroup.Taint,
) error {
	// Build updated taints: add the deprecation taint to existing taints.
	existingTaints := nodePoolTaintsToNodeGroup(nodePool.GetTaints())
	updatedTaints := nodegroup.AddTaint(existingTaints, taint)

	req := &dplanemgrv1.UpdateNodePoolRequest{
		DataPlaneId: dataplaneName,
		NodePoolId:  nodePool.GetName(),
		NodePool: &dplanemgrv1.NodePool{
			// CRITICAL: Preserve existing labels (Replace semantics).
			Labels: nodePool.GetLabels(),
			Taints: nodeGroupTaintsToNodePool(updatedTaints),
		},
	}

	_, err := a.dataplaneServiceClient.UpdateNodePool(ctx, req)
	if err != nil {
		tilog.Error(ctx, "failed to update nodepool taint",
			zap.String("dataplaneName", dataplaneName),
			zap.String("nodePoolName", nodePool.GetName()),
			zap.String("taintKey", taint.Key),
			zap.String("taintValue", taint.Value),
			zap.Error(err))
		return status.Errorf(codes.Internal, "failed to update nodepool %s taint: %v", nodePool.GetName(), err)
	}

	tilog.Info(ctx, "applied taint to nodepool",
		zap.String("dataplaneName", dataplaneName),
		zap.String("nodePoolName", nodePool.GetName()),
		zap.String("taintKey", taint.Key),
		zap.String("taintValue", taint.Value))
	return nil
}