package cron

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	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/pkg/dataplanemgr"
	"github.com/tidbcloud/cluster-service-ng/pkg/nodegroup"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const retiringProcessorLockKey = "retiring_processor_job"

// RetiringProcessorJob is a cron job that processes RETIRING node instance groups.
// For each RETIRING group, it checks every dataplane to determine if nodepools belonging
// to the group are still in use (currentReplicas > 0). If a dataplane's nodepools are all
// unused, it deletes those nodepools and sets the dataplane override status to RETIRED.
// When all dataplanes are RETIRED, it automatically sets the region-level group status to RETIRED.
//
// Uses a MySQL advisory lock to prevent concurrent executions across replicas,
// plus a local sync.Mutex to avoid overlapping within the same process.
type RetiringProcessorJob struct {
	nodeGroupRepo          repo.NodeInstanceGroupRepo
	dataplaneNodeGroupRepo repo.DataplaneNodeInstanceGroupRepo
	dpmClient              dataplanemgr.Client
	lockRepo               repo.LockRepo
	expression             string
	running                sync.Mutex
	replicasZeroSince      map[string]time.Time // in-memory tracking: key="dp/spec/instance" -> first-zero timestamp
}

// NewRetiringProcessorJob creates a new RetiringProcessorJob with all dependencies.
func NewRetiringProcessorJob(
	nodeGroupRepo repo.NodeInstanceGroupRepo,
	dataplaneNodeGroupRepo repo.DataplaneNodeInstanceGroupRepo,
	dpmClient dataplanemgr.Client,
	lockRepo repo.LockRepo,
	expression string,
) *RetiringProcessorJob {
	return &RetiringProcessorJob{
		nodeGroupRepo:          nodeGroupRepo,
		dataplaneNodeGroupRepo: dataplaneNodeGroupRepo,
		dpmClient:              dpmClient,
		lockRepo:               lockRepo,
		expression:             expression,
		replicasZeroSince:      make(map[string]time.Time),
	}
}

// Name returns the job name for identification.
func (j *RetiringProcessorJob) Name() string {
	return "RetiringProcessorJob"
}

// CronExpression returns the cron schedule for this job.
func (j *RetiringProcessorJob) CronExpression() string {
	return j.expression
}

// Execute runs the main job logic: find all RETIRING groups and process each one.
func (j *RetiringProcessorJob) Execute(ctx context.Context) error {
	if !j.running.TryLock() {
		tilog.Warn(ctx, "retiring processor job still running, skipping this execution")
		return nil
	}
	defer j.running.Unlock()

	// Acquire distributed lock to prevent concurrent executions across replicas.
	unlock, err := j.lockRepo.Lock(ctx, &entity.Lock{Key: retiringProcessorLockKey, TimeoutSecs: 0})
	if err != nil {
		tilog.Warn(ctx, "failed to acquire distributed lock for retiring processor, skipping", zap.Error(err))
		return nil
	}
	defer func() {
		if unlockErr := unlock(); unlockErr != nil {
			tilog.Warn(ctx, "failed to release distributed lock for retiring processor", zap.Error(unlockErr))
		}
	}()

	tilog.Info(ctx, "retiring processor job started")

	// Load all RETIRING node instance groups.
	retiringGroups, err := j.nodeGroupRepo.ListByStatus(ctx, nodegroup.NodeInstanceGroupStatusRetiring)
	if err != nil {
		tilog.Error(ctx, "failed to list retiring node instance groups", zap.Error(err))
		return fmt.Errorf("retiring processor job failed: %w", err)
	}

	if len(retiringGroups) == 0 {
		tilog.Info(ctx, "no retiring node instance groups found")
		return nil
	}

	// List all dataplanes.
	resp, err := j.dpmClient.ListDataPlanes(ctx, &dplanemgrv1.ListDataPlanesRequest{})
	if err != nil {
		tilog.Error(ctx, "failed to list dataplanes", zap.Error(err))
		return fmt.Errorf("retiring processor job failed: %w", err)
	}

	dataplaneNames := make([]string, 0, len(resp.GetDataPlanes()))
	for _, dp := range resp.GetDataPlanes() {
		if dp != nil && dp.GetId() != "" {
			dataplaneNames = append(dataplaneNames, dp.GetId())
		}
	}

	for _, group := range retiringGroups {
		if group == nil {
			continue
		}
		if err := j.processRetiringGroup(ctx, group, dataplaneNames); err != nil {
			nodeSpec := entity.GetComponentNodeSpec(group.GetNodeSpec())
			primaryInstance := group.GetPrimaryInstanceType()
			tilog.Error(ctx, "error processing retiring group",
				zap.String("nodeGroupName", group.GetName()),
				zap.String("nodeSpec", nodeSpec),
				zap.String("primaryInstance", primaryInstance),
				zap.Error(err))
			continue
		}
	}

	tilog.Info(ctx, "retiring processor job completed")
	return nil
}

// processRetiringGroup processes a single RETIRING node instance group across all dataplanes.
func (j *RetiringProcessorJob) processRetiringGroup(ctx context.Context, group *entity.NodeInstanceGroup, dataplaneNames []string) error {
	nodeSpec := entity.GetComponentNodeSpec(group.GetNodeSpec())
	primaryInstance := group.GetPrimaryInstanceType()

	tilog.Info(ctx, "processing retiring group",
		zap.String("nodeSpec", nodeSpec),
		zap.String("primaryInstance", primaryInstance))

	// Collect all instance types belonging to this group (primary + fallbacks).
	instanceTypes := collectInstanceTypes(group)

	allRetired := true

	for _, dataplaneName := range dataplaneNames {
		// Load the dataplane override for this group.
		override, err := j.dataplaneNodeGroupRepo.LoadByDataplaneAndSpec(ctx, dataplaneName, nodeSpec, primaryInstance)
		if err != nil {
			return fmt.Errorf("failed to load dataplane override for %s: %w", dataplaneName, err)
		}

		// Skip dataplanes that are already RETIRED.
		if override != nil && override.GetStatus() == apiv1.DataplaneNodeInstanceGroup_RETIRED {
			continue
		}

		// List nodepools once and reuse for both the active-check and cleanup.
		nodePools, err := j.listNodePools(ctx, dataplaneName)
		if err != nil {
			tilog.Error(ctx, "failed to list nodepools for dataplane",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.Error(err))
			allRetired = false
			continue
		}

		matchingPools, err := j.filterMatchingNodePools(nodePools, nodeSpec, instanceTypes)
		if err != nil {
			tilog.Error(ctx, "failed to parse nodeSpec for retiring group",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.Error(err))
			allRetired = false
			continue
		}

		// If no matching nodepools exist, skip the confirmation window and proceed
		// directly to RETIRED — no CurrentReplicas reporting-delay risk when pools don't exist.
		if len(matchingPools) == 0 {
			tilog.Info(ctx, "no matching nodepools found for retiring group on dataplane, setting RETIRED",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.String("primaryInstance", primaryInstance))
			// Clean up any stale zero-since cache entry to prevent incorrect confirmation in future cycles.
			delete(j.replicasZeroSince, j.zeroSinceKey(dataplaneName, nodeSpec, primaryInstance))
			if err := j.setDataplaneOverrideRetired(ctx, override, dataplaneName, nodeSpec, primaryInstance); err != nil {
				tilog.Error(ctx, "failed to set dataplane override to RETIRED",
					zap.String("dataplaneName", dataplaneName),
					zap.String("nodeSpec", nodeSpec),
					zap.Error(err))
				allRetired = false
				continue
			}
			continue
		}

		// Check if any matching nodepool still has active replicas.
		hasActive := false
		for _, np := range matchingPools {
			if np.GetCurrentReplicas() > 0 {
				hasActive = true
				tilog.Info(ctx, "found active nodepool for retiring group on dataplane",
					zap.String("dataplaneName", dataplaneName),
					zap.String("nodeSpec", nodeSpec),
					zap.String("primaryInstance", primaryInstance),
					zap.String("nodepoolName", np.GetName()),
					zap.Uint32("currentReplicas", np.GetCurrentReplicas()))
				break
			}
		}
		if hasActive {
			// Replicas are non-zero, clear in-memory zero-since timestamp
			delete(j.replicasZeroSince, j.zeroSinceKey(dataplaneName, nodeSpec, primaryInstance))
			allRetired = false
			continue
		}

		// All nodepools for this group on this dataplane have CurrentReplicas=0.
		// Because DPM's CurrentReplicas can have reporting delays, require 10 minutes
		// of sustained zero before proceeding with deletion.
		key := j.zeroSinceKey(dataplaneName, nodeSpec, primaryInstance)
		zeroSince, ok := j.replicasZeroSince[key]
		if !ok {
			// First time seeing all zeros, record the timestamp in memory
			j.replicasZeroSince[key] = time.Now()
			tilog.Info(ctx, "nodepools first observed as zero replicas, waiting for confirmation",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.String("primaryInstance", primaryInstance))
			allRetired = false
			continue
		}
		if time.Since(zeroSince) < ReplicasZeroConfirmDuration {
			tilog.Info(ctx, "nodepools zero but confirmation period not elapsed",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.Duration("elapsed", time.Since(zeroSince)),
				zap.Duration("required", ReplicasZeroConfirmDuration))
			allRetired = false
			continue
		}

		// Confirmed zero for required duration — delete nodepools and set RETIRED.
		if err := j.deleteNodePools(ctx, dataplaneName, matchingPools); err != nil {
			tilog.Error(ctx, "failed to cleanup nodepools for dataplane",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.Error(err))
			allRetired = false
			continue
		}

		// Set the dataplane override to RETIRED.
		if err := j.setDataplaneOverrideRetired(ctx, override, dataplaneName, nodeSpec, primaryInstance); err != nil {
			tilog.Error(ctx, "failed to set dataplane override to RETIRED",
				zap.String("dataplaneName", dataplaneName),
				zap.String("nodeSpec", nodeSpec),
				zap.Error(err))
			allRetired = false
			continue
		}

		// Clean up in-memory zero-since cache for this dataplane.
		delete(j.replicasZeroSince, key)

		tilog.Info(ctx, "dataplane marked RETIRED for retiring group",
			zap.String("dataplaneName", dataplaneName),
			zap.String("nodeSpec", nodeSpec),
			zap.String("primaryInstance", primaryInstance))
	}

	if !allRetired {
		tilog.Info(ctx, "not all dataplanes are retired, keeping group in RETIRING state",
			zap.String("nodeSpec", nodeSpec),
			zap.String("primaryInstance", primaryInstance))
		return nil
	}

	// All dataplanes are RETIRED — set region-level group to RETIRED.
	tilog.Info(ctx, "all dataplanes retired, setting group to RETIRED",
		zap.String("nodeSpec", nodeSpec),
		zap.String("primaryInstance", primaryInstance))

	group.Status = apiv1.NodeInstanceGroup_RETIRED
	if err := j.nodeGroupRepo.Save(ctx, group); err != nil {
		return fmt.Errorf("failed to update node instance group to RETIRED: %w", err)
	}

	tilog.Info(ctx, "node instance group set to RETIRED",
		zap.String("nodeSpec", nodeSpec),
		zap.String("primaryInstance", primaryInstance))

	return nil
}

// zeroSinceKey returns the composite key for in-memory replicasZeroSince tracking.
func (j *RetiringProcessorJob) zeroSinceKey(dataplaneName, nodeSpec, primaryInstance string) string {
	return dataplaneName + "/" + nodeSpec + "/" + primaryInstance
}

// filterMatchingNodePools returns nodepools that belong to the node instance group.
func (j *RetiringProcessorJob) filterMatchingNodePools(nodePools []*dplanemgrv1.NodePool, nodeSpec string, instanceTypes map[string]nodegroup.NodePoolRole) ([]*dplanemgrv1.NodePool, error) {
	cpu, mem, err := nodegroup.ParseSpec(nodeSpec)
	if err != nil {
		return nil, fmt.Errorf("failed to parse nodeSpec %q: %w", nodeSpec, err)
	}

	var matched []*dplanemgrv1.NodePool
	for _, np := range nodePools {
		if np == nil {
			continue
		}
		if matchesNodeGroup(np, nodeSpec, cpu, mem, instanceTypes) {
			matched = append(matched, np)
		}
	}
	return matched, nil
}

// deleteNodePools deletes the given nodepools on a dataplane.
// NotFound errors are treated as success (idempotency).
func (j *RetiringProcessorJob) deleteNodePools(ctx context.Context, dataplaneName string, nodePools []*dplanemgrv1.NodePool) error {
	for _, np := range nodePools {
		tilog.Info(ctx, "deleting nodepool for retiring group",
			zap.String("dataplaneName", dataplaneName),
			zap.String("nodepoolName", np.GetName()))

		if _, err := j.dpmClient.DeleteNodePool(ctx, &dplanemgrv1.DeleteNodePoolRequest{
			DataPlaneId: dataplaneName,
			NodePoolId:  np.GetName(),
		}); err != nil {
			// Treat NotFound as success — another replica or previous run may have already deleted it.
			if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
				tilog.Info(ctx, "nodepool already deleted, skipping",
					zap.String("dataplaneName", dataplaneName),
					zap.String("nodepoolName", np.GetName()))
				continue
			}
			return fmt.Errorf("failed to delete nodepool %s on dataplane %s: %w", np.GetName(), dataplaneName, err)
		}
	}

	return nil
}

// setDataplaneOverrideRetired updates the already-loaded override to RETIRED status.
// If override is nil (no existing record), a new one is created.
func (j *RetiringProcessorJob) setDataplaneOverrideRetired(ctx context.Context, override *entity.DataplaneNodeInstanceGroup, dataplaneName, nodeSpec, primaryInstance string) error {
	if override == nil {
		cpu, mem, parseErr := nodegroup.ParseSpec(nodeSpec)
		if parseErr != nil {
			return fmt.Errorf("failed to parse node spec: %w", parseErr)
		}
		override = &entity.DataplaneNodeInstanceGroup{
			DataplaneName:       dataplaneName,
			NodeSpec:            &apiv1.NodeSpec{Cpu: cpu, MemoryGi: mem},
			PrimaryInstanceType: primaryInstance,
		}
	}

	override.Status = apiv1.DataplaneNodeInstanceGroup_RETIRED
	if err := j.dataplaneNodeGroupRepo.Save(ctx, override); err != nil {
		return fmt.Errorf("failed to save dataplane override as RETIRED: %w", err)
	}

	return nil
}

// listNodePools lists all nodepools for a dataplane.
func (j *RetiringProcessorJob) listNodePools(ctx context.Context, dataplaneName string) ([]*dplanemgrv1.NodePool, error) {
	resp, err := j.dpmClient.ListNodePools(ctx, &dplanemgrv1.ListNodePoolsRequest{
		DataPlaneId: dataplaneName,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to list nodepools for dataplane %s: %w", dataplaneName, err)
	}
	return resp.GetNodePools(), nil
}

// matchesNodeGroup checks if a nodepool belongs to the node instance group.
// It matches by either:
// - InstanceTypeManager-managed label (tidbcloud.com/node-spec) + instance type in the group
// - Dataplane default pool: CPU/Memory match + instance type in the group
func matchesNodeGroup(np *dplanemgrv1.NodePool, nodeSpec string, cpu, mem uint32, instanceTypes map[string]nodegroup.NodePoolRole) bool {
	labels := np.GetLabels()
	isManaged := nodegroup.IsManagedByInstanceTypeManager(labels)

	if isManaged {
		if !strings.EqualFold(labels[nodegroup.LabelNodeSpec], nodeSpec) {
			return false
		}
	} else {
		if np.GetCpuCore() != cpu || np.GetMemoryGi() != mem {
			return false
		}
	}

	instanceType := np.GetInstanceType()
	expectedRole, ok := instanceTypes[instanceType]
	if !ok {
		return false
	}

	return nodegroup.NodePoolMatchesRole(np, expectedRole)
}

// collectInstanceTypes returns a set of all instance types (primary + fallbacks) for a group.
func collectInstanceTypes(group *entity.NodeInstanceGroup) map[string]nodegroup.NodePoolRole {
	types := map[string]nodegroup.NodePoolRole{
		group.GetPrimaryInstanceType(): nodegroup.NodePoolRolePrimary,
	}
	for _, fb := range group.GetFallbackInstances() {
		if fb != nil && fb.GetInstanceType() != "" {
			if _, exists := types[fb.GetInstanceType()]; exists {
				continue
			}
			types[fb.GetInstanceType()] = nodegroup.NodePoolRoleFallback
		}
	}
	return types
}