package task

import (
	"context"
	"fmt"
	"time"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"gopkg.in/yaml.v3"

	k8sent "github.com/tidbcloud/devops/entity/k8supgrade"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/infra/utils/container"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/generator"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/infrainfo"
)

// syncPendingTask syncs the pending task.
// Generally, it will only sync a pending task once.
func (s *svc) syncPendingTask(ctx context.Context, taskID uint64) {
	logger := log.FromContext(ctx, "func", "syncPendingTask")

	task, err := s.taskRepo.GetTask(ctx, taskID, true)
	if err != nil {
		s.Logger(ctx).Error(err, "failed to get the latest task")
		return
	}

	// Record DB-state signatures BEFORE any sync processing.
	// At this point Info is nil for every cluster, so PersistenceSignature uses
	// K8sSnapshot (the value stored in DB) as the effective snapshot — which is
	// exactly the baseline we want to diff against after sync.
	dbSigs := make(map[string]string, len(task.K8sClusterList)+len(task.ExcludedK8sClusters))
	for _, c := range task.K8sClusterList {
		dbSigs[c.Name] = c.Signature()
	}
	for _, c := range task.ExcludedK8sClusters {
		dbSigs[c.Name] = c.Signature()
	}

	selector, err := getSelector(task)
	if err != nil {
		logger.Error(err, "failed to build selector for task", "task", taskID)
		return
	}
	included, excluded, err := s.getIncludedAndExcludedK8sClusters(ctx, task, selector)
	if err != nil {
		logger.Error(err, "failed to get k8s clusters for pending task")
		return
	}

	regions := make(map[string]int64)
	needResync := false
	// TODO: use goroutines to speed up the process
	for _, cluster := range included {
		s.syncTenantRegionInfo(regions, cluster.BusinessRegion, true)

		if cluster.Info == nil {
			cluster.Info, err = s.infraInfoSvc.GetK8sInformation(ctx, &infrainfo.GetK8SInformationReq{
				CloudProvider: cluster.CloudProvider,
				Name:          cluster.Name,
				Region:        cluster.Region,
			})
			if err != nil {
				logger.Error(err, "failed to get k8s information", "cluster", cluster.Name)
				continue
			}
		}

		if _, err = s.getUpgradeOperation(ctx, k8sent.NewReadOnlyTask(task), cluster, time.Unix(task.Deadline, 0)); err != nil {
			logger.Error(err, "failed to generate the UpgradeOperation CR", "cluster", cluster.Name)
			needResync = true
			continue
		}
	}

	// Collect only clusters whose persisted fields actually changed in this sync
	// pass to avoid unnecessary DB writes.
	//
	// Rules:
	//   ID==0 : newly discovered → always write
	//   ID>0  : write only when signature differs from DB baseline
	var dirty []*k8sent.K8sCluster
	for _, c := range included {
		if c.ID == 0 || c.Signature() != dbSigs[c.Name] {
			dirty = append(dirty, c)
		}
	}
	for _, c := range excluded {
		if c.ID == 0 || c.Signature() != dbSigs[c.Name] {
			dirty = append(dirty, c)
		}
	}

	if len(dirty) > 0 {
		conflicted, upsertErr := s.taskRepo.UpsertK8sClustersWithVersionGuard(ctx, dirty)
		if upsertErr != nil {
			logger.Error(upsertErr, "failed to upsert k8s clusters")
			return
		}
		if len(conflicted) > 0 {
			// Some clusters were modified concurrently; the stale snapshot was not
			// written back. Schedule a resync so we pick up the latest state.
			logger.Info("version conflict detected, scheduling resync", "conflicted_clusters", conflicted)
			needResync = true
		}
	}

	if err = s.taskRepo.PatchTask(ctx, task, func(current *k8sent.Task) error {
		current.Regions = regions
		return nil
	}); err != nil {
		logger.Error(err, "failed to patch task")
		needResync = true
	}

	if needResync {
		s.syncTasksChan <- task.ID
	}
}

func getSelector(task *k8sent.Task) (k8sent.Selector, error) {
	return k8sent.SelectorFromTask(task)
}

// getIncludedAndExcludedK8sClusters gets the k8s clusters need to be upgraded for the active task.
// Note that if an included cluster is deleted (not found in the listed k8s information),
// it's still considered as included.
func (s *svc) getIncludedAndExcludedK8sClusters(ctx context.Context, task *k8sent.Task, selector k8sent.Selector) (included, excluded map[string]*k8sent.K8sCluster, err error) {
	logger := log.FromContext(ctx, "func", "getIncludedAndExcludedK8sClusters")

	req := &infrainfo.ListK8SInformationReq{
		CloudProvider: task.CloudProvider,
		Version:       task.VersionToBeUpgraded,
	}
	infoList, err := s.infraInfoSvc.ListK8SInformation(ctx, req)
	if err != nil {
		logger.Error(err, "failed to list k8s information", "req", req)
		return included, excluded, err
	}
	logger.Info("list k8s information for the pending task", "count", len(infoList.Items))

	included = container.SliceToMap(task.K8sClusterList, func(c *k8sent.K8sCluster) string { return c.Name })
	excluded = container.SliceToMap(task.ExcludedK8sClusters, func(c *k8sent.K8sCluster) string { return c.Name })
	for _, info := range infoList.Items {
		// `ListK8SInformation` API does not contain tidb clusters' info,
		//  so we need to call `GetK8sInformation` to get them.
		detail, err := s.infraInfoSvc.GetK8sInformation(ctx, &infrainfo.GetK8SInformationReq{
			CloudProvider: info.CloudProvider,
			Name:          info.K8sName,
			Region:        info.Region,
		})
		if err != nil {
			logger.Error(err, "failed to get detail k8s information", "cluster", info.K8sName)
			continue
		}
		logger.Info("get detail k8s information", "detail", detail)

		if !selector.Matches(ctx, detail) {
			cluster, ok := included[info.K8sName]
			if ok {
				logger.Info("exclude included k8s cluster since it does not match the selector", "cluster", info.K8sName, "selector", selector.Name())
				delete(included, info.K8sName)
			}
			if _, ok := excluded[info.K8sName]; !ok {
				if cluster != nil {
					excluded[info.K8sName] = cluster
					excluded[info.K8sName].IsExcluded = true
				} else {
					excluded[info.K8sName] = &k8sent.K8sCluster{
						Name:          info.K8sName,
						CloudProvider: info.CloudProvider,
						TaskID:        task.ID,
						IsExcluded:    true,
					}
				}
				logger.Info("exclude k8s cluster since it does not match the selector", "cluster", info.K8sName, "selector", selector.Name())
			}
		} else {
			logger.Info("match the selector", "cluster", info.K8sName, "selector", selector.Name())
		}

		if _, ok := excluded[info.K8sName]; ok {
			logger.V(5).Info("skip excluded k8s cluster", "cluster", info.K8sName)
			continue
		}

		if _, ok := included[info.K8sName]; !ok {
			included[info.K8sName] = &k8sent.K8sCluster{
				Name:           info.K8sName,
				CloudProvider:  info.CloudProvider,
				TaskID:         task.ID,
				IsExcluded:     false,
				TenantName:     detail.TenantName,
				TenantKind:     detail.TenantKind,
				BusinessRegion: detail.BusinessRegion,
				Region:         info.Region,
			}
		}
		included[info.K8sName].Info = detail
	}

	return included, excluded, nil
}

func (s *svc) getUpgradeOperation(ctx context.Context, task k8sent.ReadOnlyTask, cluster *k8sent.K8sCluster, schedule time.Time) (uo *k8sent.UpgradeOperation, err error) {
	logger := log.FromContext(ctx, "func", "getUpgradeOperation")
	if cluster.Info == nil {
		return nil, fmt.Errorf("k8s information of %s is nil", cluster.Name)
	}
	var desiredVersion string
	uo = new(k8sent.UpgradeOperation)
	if !cluster.DisableAutoGeneration {
		// Default behavior: auto-generate the UpgradeOperation CR only once after task creation,
		// then disable auto generation to avoid overwriting steps unexpectedly.
		// Users can re-enable auto generation later via UpdateK8sClusterAutoGeneration(false).
		firstGeneration := cluster.UpgradeOperationCR == ""

		uoName, uoNs := s.generator.GenerateUpgradeOperationKey(cluster.Info, task)
		uo, desiredVersion = s.generator.GenerateUpgradeOperation(ctx, cluster.Info, uoName, uoNs)
		logger.Info("generate upgrade operation", "name", uoName, "namespace", uoNs, "desired_version", desiredVersion, "uo", uo)

		if firstGeneration {
			cluster.DisableAutoGeneration = true
		}
	} else {
		// If DisableAutoGeneration is true, will use the provided UpgradeOperation CR instead of the generated one.
		if err := yaml.Unmarshal([]byte(cluster.UpgradeOperationCR), uo); err != nil {
			logger.Error(err, "failed to unmarshal upgrade operation", "name", cluster.Name, "uo", cluster.UpgradeOperationCR)
			return nil, fmt.Errorf("failed to unmarshal the UpgradeOperation CR for cluster %s: %w", cluster.Name, err)
		}
		desiredVersion = generator.GetDesiredVersion(uo)
	}

	if !schedule.IsZero() {
		uo.Schedule = schedule.Format(cloudv1.UpgradeOperationScheduleLayout)
	}
	if uo.Annotations == nil {
		uo.Annotations = map[string]string{}
	}
	cluster.DesiredVersion = desiredVersion
	uoBytes, err := yaml.Marshal(*uo)
	if err != nil {
		logger.Error(err, "failed to marshal upgrade operation", "name", cluster.Name)
		return nil, fmt.Errorf("failed to marshal UpgradeOperation CR: %w", err)
	}
	cluster.UpgradeOperationCR = string(uoBytes)
	return uo, nil
}