package task

import (
	"context"
	"errors"
	"fmt"
	"runtime/debug"
	"slices"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/dig"
	"gopkg.in/yaml.v3"
	"gorm.io/gorm"
	apierrors "k8s.io/apimachinery/pkg/api/errors"

	"github.com/tidbcloud/devops/entity/event"
	k8sent "github.com/tidbcloud/devops/entity/k8supgrade"
	models "github.com/tidbcloud/devops/models/k8supgrade"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/clients/larkcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/larkcli/notify"
	"github.com/tidbcloud/devops/pkg/infra/config"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/infra/utils/repoopt"
	"github.com/tidbcloud/devops/pkg/service"
	apprapi "github.com/tidbcloud/devops/pkg/service/approval/api"
	adsvc "github.com/tidbcloud/devops/pkg/service/audit"
	cmxapi "github.com/tidbcloud/devops/pkg/service/clustermaintenance/api"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/generator"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/infrainfo"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/task/action"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/task/api"
	"github.com/tidbcloud/devops/pkg/service/k8supgrade/task/repo"
	msgapi "github.com/tidbcloud/devops/pkg/service/message/api"
	tenantsvc "github.com/tidbcloud/devops/pkg/service/tenant"
)

const (
	svcName               = "K8sUpgradeTaskService"
	defaultSyncInterval   = 15 * time.Minute
	defaultSyncChanSize   = 20
	defaultMaxSyncWorkers = 10
)

type ServiceParam struct {
	dig.In

	Conf                 *config.RootConfig
	Logger               logr.Logger
	DBProvider           db.TransProvider
	InfraInfoSvc         infrainfo.ServiceAPI
	ApprovalService      apprapi.ApprovalServiceAPI
	AuditService         adsvc.AuditServiceAPI
	MaintenanceWindowSvc cmxapi.ServiceAPI
	TenantSvc            tenantsvc.TenantServiceAPI
	MessageSvc           msgapi.MessageServiceAPI
	LarkClient           *larkcli.LarkClient
	CoreSvcCli           *coresvccli.CoreSvcCli
}

type ServiceResult struct {
	dig.Out

	API     api.API
	Service service.Service `group:"service"`
}

type svc struct {
	*service.BaseService

	conf                 config.K8sUpgradeServiceConfig
	taskRepo             repo.Repo
	dbp                  db.TransProvider
	infraInfoSvc         infrainfo.ServiceAPI
	apprSvc              apprapi.ApprovalServiceAPI
	auditSvc             adsvc.AuditServiceAPI
	generator            generator.API
	maintenanceWindowSvc cmxapi.ServiceAPI
	tenantSvc            tenantsvc.TenantServiceAPI
	emailSvc             msgapi.EmailServiceAPI
	larkCli              *larkcli.LarkClient
	syncTasksChan        chan uint64
	// inflightTasks tracks tasks currently being processed by a worker.
	// It prevents the same task from being handled concurrently within this process.
	inflightTasks sync.Map
}

var _ api.API = &svc{}

func (s *svc) ListSupportedEOSVersions(_ context.Context) (map[string][]string, error) {
	out := make(map[string][]string, len(s.conf.SupportedEOSVersions))
	for k, v := range s.conf.SupportedEOSVersions {
		cp := make([]string, 0, len(v))
		cp = append(cp, v...)
		out[k] = cp
	}
	return out, nil
}

func (s *svc) IsK8sUpgradeAdmin(_ context.Context, email string) bool {
	return slices.Contains(s.conf.UpgradePlanConfig.AdminApprovers, email)
}

func NewService(p ServiceParam) (ServiceResult, error) {
	base := service.NewBaseService(svcName, service.WithDummy(p.Conf.Services.K8sUpgrade.Dummy))
	if p.Conf.Services.K8sUpgrade.Dummy {
		return ServiceResult{
			Service: base,
			API:     &api.DummyAPI{},
		}, nil
	}

	s := &svc{
		BaseService:          base,
		conf:                 p.Conf.Services.K8sUpgrade,
		taskRepo:             repo.NewRepo(p.DBProvider),
		dbp:                  p.DBProvider,
		infraInfoSvc:         p.InfraInfoSvc,
		apprSvc:              p.ApprovalService,
		auditSvc:             p.AuditService,
		generator:            generator.New(p.Conf),
		maintenanceWindowSvc: p.MaintenanceWindowSvc,
		tenantSvc:            p.TenantSvc,
		emailSvc:             p.MessageSvc,
		larkCli:              p.LarkClient,
		syncTasksChan:        make(chan uint64, defaultSyncChanSize),
	}

	return ServiceResult{Service: s, API: s}, nil
}

func (s *svc) Run(ctx context.Context) (err error) {
	defer func() {
		if r := recover(); r != nil {
			s.Logger(ctx).Error(fmt.Errorf("panic: %v", r), "recovered from panic")
			debug.PrintStack()
			err = fmt.Errorf("panic: %v", r)
		}
	}()

	interval := defaultSyncInterval
	if s.conf.SyncUpgradeTasksInterval != nil {
		interval = *s.conf.SyncUpgradeTasksInterval
	}
	s.Logger(ctx).Info("sync k8s upgrade tasks regularly", "interval", interval.String())
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	var wg sync.WaitGroup
	defer wg.Wait()
	for i := 0; i < defaultMaxSyncWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for task := range s.syncTasksChan {
				s.handleTask(ctx, task)
			}
		}()
	}

	for {
		select {
		case <-ctx.Done():
			s.Logger(ctx).Info("exit because context canceled", "cause", context.Cause(ctx))
			close(s.syncTasksChan)
			return ctx.Err()

		case <-ticker.C:
			// List all tasks with status approving and running regularly.
			tasks, err := s.taskRepo.ListTasks(ctx,
				repo.StatusIn([]k8sent.TaskStatus{k8sent.TaskStatusPending, k8sent.TaskStatusApproving, k8sent.TaskStatusRunning}),
				// Only select task ID to reduce the data size.
				repo.SelectIDOnly(),
			)
			if err != nil {
				s.Logger(ctx).Error(err, "failed to list approving, and running tasks")
				continue
			}
			s.Logger(ctx).Info("list k8s upgrade tasks", "count", len(tasks))

			for _, task := range tasks {
				select {
				case s.syncTasksChan <- task.ID:
				default:
					s.Logger(ctx).Error(fmt.Errorf("syncTasksChan is full, skip task %d", task.ID), "syncTasksChan")
				}
			}
		}
	}
}

func (s *svc) handleTask(ctx context.Context, taskID uint64) {
	// Prevent concurrent processing of the same task within this process.
	// Cross-process safety is ensured by optimistic locking in the repo layer.
	if _, loaded := s.inflightTasks.LoadOrStore(taskID, struct{}{}); loaded {
		s.Logger(ctx).Info("task is already being processed, skipping", "task_id", taskID)
		return
	}
	defer s.inflightTasks.Delete(taskID)

	syncCtx := log.NewContext(ctx, s.Logger(ctx), "task_id", taskID)
	// get the latest task to get the latest status
	currentTask, err := s.taskRepo.GetTask(syncCtx, taskID, false)
	if err != nil {
		s.Logger(ctx).Error(err, "failed to get the latest task")
		return
	}
	s.Logger(ctx).Info("handle k8s upgrade task", "task_id", taskID, "task_status", currentTask.Status)

	switch currentTask.Status {
	case k8sent.TaskStatusPending:
		s.syncPendingTask(syncCtx, taskID)

	case k8sent.TaskStatusApproving:
		s.syncApprovingTask(syncCtx, currentTask)

	case k8sent.TaskStatusRunning:
		s.syncRunningTask(syncCtx, taskID)

	case k8sent.TaskStatusCancelled:
		s.syncCanceledTask(syncCtx, taskID)
	}
}

// syncTenantRegionInfo syncs the task's regions info.
// A task contains multiple k8s clusters.
// Adding or excluding a k8s cluster will cause the task's regions info to change.
func (s *svc) syncTenantRegionInfo(regions map[string]int64, bizRegion string, add bool) {
	if add {
		if _, ok := regions[bizRegion]; !ok {
			regions[bizRegion] = 1
		} else {
			regions[bizRegion]++
		}
	} else {
		if _, ok := regions[bizRegion]; ok {
			regions[bizRegion]--
		}
	}
}

func (s *svc) taskComps(ctx context.Context) action.Components {
	return action.Components{
		Logger:          log.FromContext(ctx),
		Conf:            s.conf,
		Repo:            s.taskRepo,
		ApprovalService: s.apprSvc,
	}
}

func (s *svc) CreateTask(ctx context.Context, param api.CreateTaskParam) (uint64, error) {
	taskEnt := &k8sent.Task{
		Name:                param.Name,
		VersionToBeUpgraded: param.EOSVersion,
		CloudProvider:       param.CloudProvider,
		Deadline:            param.Deadline,
		MaxRestartTimes:     param.MaxRestartTimes,
		Applicant:           param.Applicant,
		Status:              k8sent.TaskStatusPending,
		SelectorType:        param.SelectorType,
		SelectorValue:       param.SelectorValue,
	}
	task, err := s.taskRepo.CreateTask(ctx, taskEnt)
	if err != nil {
		s.Logger(ctx).Error(err, "create task error when accessing database")
		return 0, err
	}
	s.syncTasksChan <- task.ID
	return task.ID, nil
}

func (s *svc) ListTasks(ctx context.Context, param api.ListTasksParam) ([]*k8sent.Task, int64, error) {
	var opts []repoopt.Option
	if len(param.CloudProviders) != 0 {
		opts = append(opts, repo.CloudProviderIn(param.CloudProviders))
	}
	if len(param.EOSVersions) != 0 {
		opts = append(opts, repo.VersionIn(param.EOSVersions))
	}
	if len(param.Status) != 0 {
		opts = append(opts, repo.StatusIn(param.Status))
	}
	total, err := s.taskRepo.CountTasks(ctx, opts...)
	if err != nil {
		return nil, 0, err
	}

	if param.Page > 0 {
		opts = append(opts, repo.WithPager(param.Page, param.PerPage))
	}
	tasks, err := s.taskRepo.ListTasks(ctx, opts...)
	if err != nil {
		return nil, 0, err
	}
	return tasks, total, nil
}

func (s *svc) GetTask(ctx context.Context, param api.GetTaskParam) (*k8sent.Task, error) {
	task, err := s.taskRepo.GetTaskWithOpts(ctx, param.TaskID, repo.GetTaskOpts{
		WithK8sClusters:         true,
		WithExcludedK8sClusters: param.WithExcludedK8sClusters,
	})
	if err != nil {
		return nil, err
	}
	s.enrichTaskClustersTenantContact(ctx, task)
	return task, nil
}

func (s *svc) CloseTask(ctx context.Context, id uint64) error {
	task, err := s.taskRepo.GetTask(ctx, id, false)
	if err != nil {
		return err
	}

	actor := action.NewK8sUpgradeTaskActor(ctx, s.taskComps(ctx), task)

	return s.dbp.Transaction(ctx, func(ctx context.Context) error {
		err := actor.Cancel()
		if err != nil {
			return err
		}

		s.syncTasksChan <- task.ID

		// record audit log
		_, err = s.buildAndRecordEvent(ctx, actor, event.K8sUpgradeTaskEventTypeCancel)
		if err != nil {
			return err
		}

		return nil
	})
}

func (s *svc) UpdateTask(ctx context.Context, taskID uint64, param api.UpdateTaskParam) error {
	err := s.taskRepo.UpdateTask(ctx, taskID, param.TaskToUpdate)
	return err
}

func (s *svc) SubmitTask(ctx context.Context, id uint64) error {
	task, err := s.taskRepo.GetTask(ctx, id, false)
	if err != nil {
		return err
	}

	actor := action.NewK8sUpgradeTaskActor(ctx, s.taskComps(ctx), task)

	return s.dbp.Transaction(ctx, func(ctx context.Context) error {
		err := actor.SendApproval()
		if err != nil {
			return err
		}

		// record audit log
		_, err = s.buildAndRecordEvent(ctx, actor, event.K8sUpgradeTaskEventTypeSendApproval)
		if err != nil {
			return err
		}

		return nil
	})
}

func (s *svc) DeleteTask(ctx context.Context, taskID uint64) error {
	task, err := s.taskRepo.GetTask(ctx, taskID, false)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil
		}
		return fmt.Errorf("get task %v failed: %w", taskID, err)
	}

	if task.Status == k8sent.TaskStatusRunning {
		return fmt.Errorf("cannot delete a running task")
	}

	if err := s.taskRepo.DeleteK8sClusters(ctx, taskID); err != nil {
		return fmt.Errorf("delete k8s clusters of task %v failed: %w", taskID, err)
	}
	return s.taskRepo.DeleteTask(ctx, taskID)
}

func (s *svc) GetK8sCluster(ctx context.Context, param api.GetK8sClusterParam) (*k8sent.K8sCluster, error) {
	cluster, err := s.taskRepo.GetK8sCluster(ctx, param.TaskID, param.K8sName)
	if err != nil {
		return nil, err
	}
	return cluster, nil
}

func (s *svc) UpdateK8sCluster(ctx context.Context, taskID uint64, name string, param api.UpdateK8sClusterParam) error {
	// If modified the UpgradeOperation CR, will disable auto generation automatically
	if len(param.Cluster.UpgradeOperationCR) > 0 {
		param.Cluster.DisableAutoGeneration = true
	}
	if err := s.taskRepo.UpdateK8sCluster(ctx, taskID, name, param.Cluster); err != nil {
		return fmt.Errorf("update k8s cluster %s of task %v failed: %w", name, taskID, err)
	}
	s.syncTasksChan <- taskID
	return nil
}

func (s *svc) UpdateK8sClusterAutoGeneration(ctx context.Context, taskID uint64, name string, disableAutoGeneration bool) error {
	err := s.dbp.WriteDB(ctx).Model(&models.K8sCluster{}).
		Where("task_id = ?", taskID).
		Where("name = ?", name).
		Updates(map[string]interface{}{
			"disable_auto_generation": disableAutoGeneration,
			"version":                 gorm.Expr("version + 1"),
		}).Error
	if err != nil {
		return fmt.Errorf("update k8s cluster %s of task %v failed: %w", name, taskID, err)
	}
	return nil
}

// ExcludeK8sCluster excludes a certain k8s cluster from a task.
// it first updates the is_excluded field of the cluster to `true`,
// then sync the region info of the task.
func (s *svc) ExcludeK8sCluster(ctx context.Context, taskID uint64, name string) error {
	err := s.dbp.WriteDB(ctx).Model(&models.K8sCluster{}).
		Where("task_id = ?", taskID).
		Where("name = ?", name).
		Updates(map[string]interface{}{
			"is_excluded": true,
			"version":     gorm.Expr("version + 1"),
		}).Error
	if err != nil {
		return fmt.Errorf("update k8s cluster %s of task %v failed: %w", name, taskID, err)
	}

	task, err := s.taskRepo.GetTask(ctx, taskID, false)
	if err != nil {
		return fmt.Errorf("get task %v failed: %w", taskID, err)
	}
	k8s, err := s.taskRepo.GetK8sCluster(ctx, taskID, name)
	if err != nil {
		return fmt.Errorf("get k8s cluster %s of task %v failed: %w", name, taskID, err)
	}

	// sync the tenant regions info of the task
	s.syncTenantRegionInfo(task.Regions, k8s.BusinessRegion, false)

	// update the task
	err = s.taskRepo.UpdateTask(ctx, taskID, task)
	if err != nil {
		return fmt.Errorf("update task %d failed: %w", taskID, err)
	}
	return nil
}

// CancelUpgradeK8sCluster cancels the upgrade of a k8s cluster.
func (s *svc) CancelUpgradeK8sCluster(ctx context.Context, taskID uint64, name string) error {
	logger := s.Logger(ctx).WithValues("task", taskID, "cluster", name, "func", "CancelUpgradeK8sCluster")

	k8s, err := s.taskRepo.GetK8sCluster(ctx, taskID, name)
	if err != nil {
		return fmt.Errorf("get k8s cluster %s of task %v failed: %w", name, taskID, err)
	}

	if err = s.dbp.WriteDB(ctx).Model(&models.K8sCluster{}).
		Where("task_id = ?", taskID).
		Where("name = ?", name).
		Updates(map[string]interface{}{
			"upgrade_operation_status": k8sent.UOStatusCancelled,
			"version":                  gorm.Expr("version + 1"),
		}).Error; err != nil {
		return fmt.Errorf("update k8s cluster %s of task %v failed: %w", name, taskID, err)
	}

	var uo k8sent.UpgradeOperation
	if err = yaml.Unmarshal([]byte(k8s.UpgradeOperationCR), &uo); err != nil {
		return fmt.Errorf("unmarshal upgrade operation CR failed: %w", err)
	}
	logger.Info("delete the UpgradeOperation CR", "name", uo.Name, "namespace", uo.Namespace)
	if err = s.infraInfoSvc.DeleteUpgradeOperation(ctx, &infrainfo.CommonUpgradeOperationReq{
		Namespace:     uo.Namespace,
		Name:          uo.Name,
		CloudProvider: k8s.CloudProvider,
		Region:        k8s.Region,
	}); err != nil && !apierrors.IsNotFound(err) {
		return fmt.Errorf("failed to delete upgrade operation: %w", err)
	}

	if err = s.larkCli.Notify.SendMessageViaWebhook(ctx, s.conf.LarkWebhook, notify.CancelUpgradeK8sClusterParam{
		K8sName: name,
		Detail:  fmt.Sprintf(s.conf.DetailLinkBaseURL, taskID),
	}); err != nil {
		return fmt.Errorf("send lark message failed: %w", err)
	}

	return nil
}