package task

import (
	"context"
	"fmt"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	maintent "github.com/tidbcloud/devops/entity/maintenance"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"

	k8sent "github.com/tidbcloud/devops/entity/k8supgrade"
	"github.com/tidbcloud/devops/pkg/infra/errs"
	"github.com/tidbcloud/devops/pkg/infra/log"
	servermetrics "github.com/tidbcloud/devops/pkg/infra/metrics/devops-server"
	cmxapi "github.com/tidbcloud/devops/pkg/service/clustermaintenance/api"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/infrainfo"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/task/utils"
)

const (
	defaultWorkers = 10
	maxWorkers     = 100

	maxBufferSize = 100
)

// syncRunningTask syncs the running task.
// It uses fan-out pattern to sync each k8s cluster.
func (s *svc) syncRunningTask(ctx context.Context, taskID uint64) {
	logger := log.FromContext(ctx, "func", "syncRunningTask")

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

	// Use bounded buffer size to avoid excessive memory usage.
	bufferSize := min(maxBufferSize, len(task.K8sClusterList))
	k8sClustersChan := make(chan *k8sent.K8sCluster, bufferSize)

	// Feed clusters to the channel in a separate goroutine to allow for backpressure
	go func() {
		defer close(k8sClustersChan)
		for _, k8s := range task.K8sClusterList {
			select {
			case k8sClustersChan <- k8s:
			case <-ctx.Done():
				return
			}
		}
	}()

	var (
		results        = make(chan *k8sent.K8sCluster, len(task.K8sClusterList))
		completedCount atomic.Uint32
		wg             sync.WaitGroup
		// Calculate appropriate worker count based on cluster count, with reasonable limits
		workerCount = min(max(defaultWorkers, len(task.K8sClusterList)/10), maxWorkers)
	)
	for i := 0; i < workerCount; i++ {
		wg.Add(1)

		go func(ctx context.Context, t k8sent.ReadOnlyTask) {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case cluster, ok := <-k8sClustersChan:
					if !ok {
						return
					}
					// Do not check if UO is completed here,
					// because the maintenance window may not be closed yet.
					if cluster.UpgradeOperationStatus == k8sent.UOStatusCancelled {
						completedCount.Add(1)
						continue
					}

					workerCtx := log.NewContext(ctx, logger, "cluster", cluster.Name)
					var err error
					cluster.Info, err = s.infraInfoSvc.GetK8sInformation(workerCtx, &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
					}
					updateMaintenanceWindowNotCreatedMetric(t.ID(), cluster)

					uoName, uoNs, currentUO, err := s.loadUpgradeOperationCR(workerCtx, t, cluster)
					if err != nil {
						logger.Error(err, "failed to get upgrade operation CR", "cluster", cluster.Name)
						continue
					}

					schedule, closed, err := s.syncMaintenanceWindow(workerCtx, t, cluster, currentUO)
					if err != nil {
						logger.Error(err, "failed to sync maintenance window", "cluster", cluster.Name)
						continue
					}

					if err = s.syncUpgradeOperation(workerCtx, t, cluster, schedule, uoName, uoNs, currentUO); err != nil {
						logger.Error(err, "failed to sync upgrade operation", "cluster", cluster.Name)
						continue
					}
					updateK8sClusterUONotStartMetric(t.ID(), cluster, schedule, closed)

					logger.Info("k8s cluster is synced", "cluster", cluster.Name, "uo_status", cluster.UpgradeOperationStatus, "mw_closed", closed)
					if closed && (cluster.UpgradeOperationStatus == string(cloudv1.UpgradeOperationPhaseCompleted) || cluster.UpgradeOperationStatus == string(cloudv1.UpgradeOperationPhaseFailed)) {
						completedCount.Add(1)
					}
					results <- cluster
				}
			}
		}(ctx, k8sent.NewReadOnlyTask(task))
	}

	// Start a goroutine to close the results channel after all other goroutines finished
	go func() {
		wg.Wait()
		close(results)
	}()

	var needUpdate []*k8sent.K8sCluster
	for cluster := range results {
		// Do not update k8s snapshots during task execution.
		cluster.Info = nil
		needUpdate = append(needUpdate, cluster)
	}

	// update k8s clusters
	if len(needUpdate) > 0 {
		if conflicted, err := s.taskRepo.UpsertK8sClustersWithVersionGuard(ctx, needUpdate); err != nil {
			logger.Error(err, "failed to update k8s clusters", "clusters", needUpdate)
			return
		} else if len(conflicted) > 0 {
			logger.Info("version conflict detected, wait for the next sync", "conflicted_clusters", conflicted)
		}
	}

	// update the task if necessary
	if int(completedCount.Load()) == len(task.K8sClusterList) {
		logger.Info("all k8s clusters have completed the upgrade operation")
		if err := s.taskRepo.PatchTask(ctx, task, func(current *k8sent.Task) error {
			current.Status = k8sent.TaskStatusCompleted
			return nil
		}); err != nil {
			logger.Error(err, "failed to patch task", "task", task.ID)
		}
	}
}

// syncMaintenanceWindow creates, starts, finishes or cancels the maintenance window.
func (s *svc) syncMaintenanceWindow(ctx context.Context, task k8sent.ReadOnlyTask, cluster *k8sent.K8sCluster, currentUO *k8sent.UpgradeOperation) (schedule time.Time, closed bool, err error) {
	logger := log.FromContext(ctx, "func", "syncMaintenanceWindow", "cluster", cluster.Name)
	deadline := task.Deadline()
	schedule = time.Unix(deadline, 0)

	if cluster.MaintenanceWindowID == nil {
		if cluster.Info.NumberOfTiDBClusters > 0 {
			logger.Info("create a maintenance window")
			param, err := utils.MakeCreateMaintenanceParam(cluster.Info, deadline)
			if err != nil {
				return schedule, closed, fmt.Errorf("failed to make create maintenance param: %w", err)
			}
			id, err := s.maintenanceWindowSvc.CreateMaintenance(ctx, param)
			if err != nil {
				logger.Error(err, "failed to create a maintenance window")
				return schedule, closed, err
			}
			cluster.MaintenanceWindowID = id
			logger.Info("created a maintenance window", "maintenance_window", *id)
		} else {
			closed = true
			logger.Info("no need to create a maintenance window since no tidb clusters")
		}
	}

	if cluster.MaintenanceWindowID != nil {
		mw, err := s.maintenanceWindowSvc.GetMaintenance(ctx, *cluster.MaintenanceWindowID, cmxapi.GetMaintenanceParam{
			WithExecutions: true,
		})
		if err != nil {
			logger.Error(err, "failed to get maintenance window")
			return schedule, closed, err
		}
		schedule = mw.Task.ScheduleTime

		if cluster.Info.NumberOfTiDBClusters == 0 && mw.Task.Status != maintent.ClusterMaintenanceTaskStatusClosed {
			// FIXME: a new cluster may be created, consider not closing the maintenance window
			logger.Info("should cancel the maintenance window when there are no tidb clusters", "maintenance_window", *cluster.MaintenanceWindowID)
			if err = utils.CloseMaintenanceWindow(ctx, s.maintenanceWindowSvc, mw); err != nil {
				logger.Error(err, "failed to close maintenance window", "maintenance_window", *cluster.MaintenanceWindowID)
			} else {
				closed = true
			}
			return schedule, closed, err
		}

		// Start or finish the maintenance window automatically
		switch mw.Task.Status {
		case maintent.ClusterMaintenanceTaskStatusRunning:

			if currentUO == nil {
				return schedule, false, nil
			}

			if IsClusterMaintenanceMode(currentUO.MaintenancePolicyMode) {
				switch cluster.UpgradeOperationStatus {
				case string(cloudv1.UpgradeOperationPhaseCompleted), string(cloudv1.UpgradeOperationPhaseFailed):
					logger.Info("close the maintenance window for the UpgradeOperation", "cluster", cluster.Name, "status", cluster.UpgradeOperationStatus)
					if err := s.closeMaintenanceWindowWithExecutions(ctx, mw); err != nil {
						logger.Error(err, "failed to close maintenance window", "maintenance_window", *cluster.MaintenanceWindowID)
					} else {
						closed = true
					}
				}
				break
			}
			// When it reaches the scheduled time, the maintenance task will be running status
			switch cluster.UpgradeOperationStatus {
			// If the UO CR is pending, we can also start the maintenance.
			// Otherwise, the restart of the tidb cluster may happen outside the maintenance window.
			case string(cloudv1.UpgradeOperationPhasePending), string(cloudv1.UpgradeOperationPhaseRunning):
				for _, exec := range mw.Task.Executions {
					if exec.Status == maintent.ClusterMaintenanceExecutionStatusPending {
						if err = s.maintenanceWindowSvc.StartMaintenanceExecution(ctx, *cluster.MaintenanceWindowID, exec.ID); err != nil {
							logger.Error(err, "failed to start maintenance execution", "maintenance_window", *cluster.MaintenanceWindowID, "execution", exec.ID)
						} else {
							logger.Info("started the maintenance execution", "maintenance_window", *cluster.MaintenanceWindowID, "execution", exec.ID)
						}
					}
				}

			case string(cloudv1.UpgradeOperationPhaseCompleted):
				// For failed UO, we need to close the window manually.
				logger.Info("close the maintenance window for the completed UpgradeOperation")
				for _, exec := range mw.Task.Executions {
					if exec.Status == maintent.ClusterMaintenanceExecutionStatusRunning {
						if err = s.maintenanceWindowSvc.FinishMaintenanceExecution(ctx, *cluster.MaintenanceWindowID, exec.ID); err != nil {
							logger.Error(err, "failed to finish maintenance execution", "maintenance_window", *cluster.MaintenanceWindowID, "execution", exec.ID)
						}
					}
				}
				if err := utils.CloseMaintenanceWindow(ctx, s.maintenanceWindowSvc, mw); err != nil {
					logger.Error(err, "failed to close maintenance window", "maintenance_window", *cluster.MaintenanceWindowID)
				} else {
					closed = true
				}
			}

		case maintent.ClusterMaintenanceTaskStatusClosed:
			closed = true
		}
	}

	return schedule, closed, nil
}

func (s *svc) closeMaintenanceWindowWithExecutions(ctx context.Context, mw *maintent.ClusterMaintenance) error {
	for _, exec := range mw.Task.Executions {
		if exec.Status == maintent.ClusterMaintenanceExecutionStatusRunning {
			if err := s.maintenanceWindowSvc.FinishMaintenanceExecution(ctx, mw.ID, exec.ID); err != nil {
				return err
			}
		}
	}
	return utils.CloseMaintenanceWindow(ctx, s.maintenanceWindowSvc, mw)
}

func (s *svc) loadUpgradeOperationCR(
	ctx context.Context,
	task k8sent.ReadOnlyTask,
	cluster *k8sent.K8sCluster,
) (uoName, uoNs string, currentUO *k8sent.UpgradeOperation, err error) {
	uoName, uoNs = s.generator.GenerateUpgradeOperationKey(cluster.Info, task)
	currentUO, err = s.infraInfoSvc.GetUpgradeOperation(ctx, &infrainfo.CommonUpgradeOperationReq{
		Name:          uoName,
		Namespace:     uoNs,
		CloudProvider: cluster.Info.CloudProvider,
		Region:        cluster.Info.Region,
	})
	if errs.IsNotFound(err) {
		return uoName, uoNs, nil, nil
	}
	if err != nil {
		return uoName, uoNs, nil, err
	}
	return uoName, uoNs, currentUO, nil
}

// syncUpgradeOperation creates or updates the UpgradeOperation CR.
func (s *svc) syncUpgradeOperation(
	ctx context.Context,
	task k8sent.ReadOnlyTask,
	cluster *k8sent.K8sCluster,
	schedule time.Time,
	uoName, uoNs string,
	currentUO *k8sent.UpgradeOperation,
) error {
	logger := log.FromContext(ctx, "func", "syncUpgradeOperation")

	if cluster.UpgradeOperationStatus == string(cloudv1.UpgradeOperationPhaseCompleted) {
		logger.Info("the UpgradeOperation is already completed and may be deleted due to TTL, no need to sync")
		return nil
	}

	uoNotFound := currentUO == nil
	if uoNotFound {
		logger.Info("the UpgradeOperation CR is not found, will create a new one", "name", uoName, "namespace", uoNs)
	}

	if shouldSyncUpgradeOperation(uoNotFound, currentUO) {
		expectUO, err := s.getUpgradeOperation(ctx, task, cluster, schedule)
		if err != nil {
			return fmt.Errorf("failed to generate the UpgradeOperation CR for %s: %w", cluster.Name, err)
		}
		if err := s.infraInfoSvc.CreateOrUpdateUpgradeOperation(ctx, &infrainfo.CommonUpgradeOperationReq{
			Name:             uoName,
			Namespace:        uoNs,
			UpgradeOperation: expectUO,
			CloudProvider:    cluster.Info.CloudProvider,
			Region:           cluster.Info.Region,
		}); err != nil {
			return fmt.Errorf("failed to create or update the UpgradeOperation CR %s/%s: %w", uoName, uoNs, err)
		}
		logger.Info("created or updated the UpgradeOperation CR", "name", uoName, "namespace", uoNs)
		cluster.UpgradeOperationStatus = string(cloudv1.UpgradeOperationPhasePending)
		return nil
	}

	if shouldSyncUOMaintenanceAnnotations(cluster, currentUO) {
		if err := s.syncUOMaintenanceAnnotations(ctx, cluster, currentUO, uoName, uoNs); err != nil {
			return fmt.Errorf("failed to sync UO maintenance annotations for %s: %w", cluster.Name, err)
		}
	}

	switch cloudv1.UpgradeOperationPhase(currentUO.Status.Phase) {
	case cloudv1.UpgradeOperationPhaseRunning, cloudv1.UpgradeOperationPhaseCompleted, cloudv1.UpgradeOperationPhaseFailed:
		logger.Info("update the UpgradeOperation CR's status", "name", uoName, "uo_namespace", uoNs, "before", cluster.UpgradeOperationStatus, "after", currentUO.Status.Phase)
		cluster.UpgradeOperationStatus = currentUO.Status.Phase
	}

	return nil
}

func shouldSyncUpgradeOperation(notFound bool, uo *k8sent.UpgradeOperation) bool {
	if notFound {
		return true
	}

	if uo != nil && uo.Status != nil {
		return uo.Status.Phase == string(cloudv1.UpgradeOperationPhaseNone) || uo.Status.Phase == string(cloudv1.UpgradeOperationPhasePending)
	}
	return false
}

func shouldSyncUOMaintenanceAnnotations(cluster *k8sent.K8sCluster, uo *k8sent.UpgradeOperation) bool {
	if cluster.MaintenanceWindowID == nil || uo == nil || uo.Status == nil {
		return false
	}
	if uo.Status.Phase != string(cloudv1.UpgradeOperationPhaseRunning) {
		return false
	}
	return IsClusterMaintenanceMode(uo.MaintenancePolicyMode)
}

func updateMaintenanceWindowNotCreatedMetric(taskID uint64, cluster *k8sent.K8sCluster) {
	labelValues := k8sClusterMetricsLabelValues(taskID, cluster)
	if shouldReportMaintenanceWindowNotCreated(cluster) {
		servermetrics.K8sUpgrade.MaintenanceWindowNotCreated.WithLabelValues(labelValues...).Set(1)
		return
	}
	servermetrics.K8sUpgrade.MaintenanceWindowNotCreated.DeleteLabelValues(labelValues...)
}

func shouldReportMaintenanceWindowNotCreated(cluster *k8sent.K8sCluster) bool {
	if cluster == nil || cluster.MaintenanceWindowID != nil {
		return false
	}
	if cluster.Info == nil {
		return false
	}
	return cluster.Info.NumberOfTiDBClusters > 0
}

func k8sClusterMetricsLabelValues(taskID uint64, cluster *k8sent.K8sCluster) []string {
	if cluster == nil {
		return []string{strconv.FormatUint(taskID, 10), "", "", ""}
	}
	return []string{
		strconv.FormatUint(taskID, 10),
		cluster.Name,
		cluster.CloudProvider,
		cluster.Region,
	}
}

func updateK8sClusterUONotStartMetric(taskID uint64, cluster *k8sent.K8sCluster, schedule time.Time, closed bool) {
	labelValues := k8sClusterMetricsLabelValues(taskID, cluster)
	if shouldReportK8sClusterUONotStart(schedule, closed, cluster) {
		servermetrics.K8sUpgrade.K8sClusterUONotStart.WithLabelValues(labelValues...).Set(1)
		return
	}
	servermetrics.K8sUpgrade.K8sClusterUONotStart.DeleteLabelValues(labelValues...)
}

func shouldReportK8sClusterUONotStart(schedule time.Time, closed bool, cluster *k8sent.K8sCluster) bool {
	if cluster == nil || cluster.MaintenanceWindowID == nil || closed || schedule.IsZero() {
		return false
	}
	if time.Now().Before(schedule) {
		return false
	}

	switch cloudv1.UpgradeOperationPhase(cluster.UpgradeOperationStatus) {
	case cloudv1.UpgradeOperationPhaseRunning, cloudv1.UpgradeOperationPhaseCompleted, cloudv1.UpgradeOperationPhaseFailed:
		return false
	default:
		return true
	}
}