package worker

import (
	"context"
	commonerrs "dataservices/commons/errs"
	"dataservices/commons/logger"
	"dataservices/commons/metrics"
	"dataservices/internal/entity"
	"dataservices/internal/repository"
	"math"
	"strconv"
	"sync"
	"time"

	worker_privatelink "dataservices/worker/commontask/privatelink"
	worker_sampledataapp "dataservices/worker/commontask/sample_dataapp"

	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

const (
	taskInitRetryInterval     = time.Second
	taskRetryIntervalMultiper = 2
	taskMaxIntervalTime       = time.Minute * 3
	taskMaxRetryTimes         = 60
)

type CommonTaskHandler func(context.Context, *entity.CommonTask) error

var commonTaskHandlers map[string]CommonTaskHandler

func init() {
	commonTaskHandlers = map[string]CommonTaskHandler{
		entity.CommonTaskTypeSampleData:  worker_sampledataapp.SampleDataappTaskHandler,
		entity.CommonTaskTypePrivatelink: worker_privatelink.PrivatelinkTaskHandler,
	}
}

type commonTaskWorker struct {
	ctx            context.Context
	jobStatusMutex sync.Mutex
	jobStatusMap   map[uint64]bool
}

func NewCommonTaskWorker(ctx context.Context) Worker {
	var worker = commonTaskWorker{
		ctx:          ctx,
		jobStatusMap: make(map[uint64]bool),
	}
	return &worker
}

func (worker *commonTaskWorker) Name() string {
	return "commontask-worker"
}

func (worker *commonTaskWorker) Handle(tracer trace.Tracer) {
	commonTasks, err := repository.SelectCommonTaskForRun(worker.ctx)
	if err != nil {
		tilog.Error(worker.ctx, "SelectCommonTaskForRun failed.", zap.Error(err))
		return
	}
	for _, taskInfo := range commonTasks {
		// Check the status of parent task to prevent concurrent execution
		if taskInfo.ParentID != 0 && taskInfo.Status == entity.CommonTaskStatusPending {
			parentTask, err := repository.SelectCommonTaskByID(worker.ctx, taskInfo.ParentID)
			if err != nil {
				logger.LogWithError(worker.ctx, err, "failed to select parent task", zap.String("task_type", taskInfo.TaskType), zap.Uint64("task_id", taskInfo.ID), zap.Uint64("parent_task_id", taskInfo.ParentID))
				continue
			}
			if !parentTask.Done() {
				logger.Info(worker.ctx, "the parent task has not been completed yet and needs to wait for a while", zap.String("task_type", taskInfo.TaskType), zap.Uint64("task_id", taskInfo.ID), zap.Uint64("parent_task_id", taskInfo.ParentID), zap.String("parent_task_status", parentTask.Status))
				continue
			}
		}
		go func(ctx context.Context, taskInfo *entity.CommonTask) {
			newctx, span := tracer.Start(ctx, "commontask_"+strconv.FormatUint(taskInfo.ID, 10))
			defer span.End()
			if err := worker.handleTask(newctx, taskInfo); err != nil {
				if !commonerrs.IsClientErr(err) {
					metrics.WorkerTaskCounter.WithLabelValues("commontask", taskInfo.TaskType, "false").Inc()
				}
			}
		}(worker.ctx, taskInfo)
	}
}

func (worker *commonTaskWorker) handleTask(ctx context.Context, taskInfo *entity.CommonTask) (err error) {

	startAt := time.Now()
	logger := tilog.WithContextL(ctx).With(
		zap.String("handler", "commontask_worker"),
		zap.Uint64("task_id", taskInfo.ID),
		zap.String("task_type", taskInfo.TaskType),
		zap.Uint64("event_id", taskInfo.EventID),
	)
	logger.Info("handleTask start...", zap.Any("task_info", taskInfo))
	defer func() {
		if perr := recover(); perr != nil {
			logger.Error("handler panicked", zap.String("cost", time.Since(startAt).String()), zap.Any("panic_err", perr), zap.Error(err))
			err = errs.NewInternalServerError("common task worker handler panicked", err)
		} else {
			logger.Info("handler completed", zap.String("cost", time.Since(startAt).String()), zap.Error(err))
		}
	}()

	// Check whether the running job exists in current id
	if worker.jobRunning(taskInfo.ID) {
		logger.Info("task already running, skip once")
		return
	}
	defer worker.jobDone(taskInfo.ID)

	logger.Info("task running", zap.Any("task_info", taskInfo))
	taskInfo.Status = entity.CommonTaskStatusRunning
	if err = repository.UpdateCommonTask(ctx, taskInfo); err != nil {
		logger.Error("UpdateCommonTask failed", zap.String("to_status", taskInfo.Status), zap.Error(err))
		return
	}
	defer func() {
		if err != nil {
			taskInfo.ErrMessage = err.Error()
			if taskInfo.RetryTimes >= taskMaxRetryTimes || errs.IsError(err, commonerrs.ErrProhibitRetry) {
				taskInfo.Status = entity.CommonTaskStatusFailed
			} else {
				taskInfo.Status = entity.CommonTaskStatusDormant
				taskInfo.RetryTimes++
				interval := math.Min(float64(taskMaxIntervalTime), float64(taskInitRetryInterval)*math.Pow(float64(taskRetryIntervalMultiper), float64(taskInfo.RetryTimes-1)))
				taskInfo.NextRetry = int(time.Now().Add(time.Duration(interval)).Unix())
			}
		} else {
			taskInfo.Status = entity.CommonTaskStatusSucceed
		}
		repository.UpdateCommonTask(ctx, taskInfo)
	}()

	if handler, ok := commonTaskHandlers[taskInfo.TaskType]; ok {
		err = handler(ctx, taskInfo)
	} else {
		logger.Warn("unsupported task type")
	}
	return
}

func (worker *commonTaskWorker) jobRunning(id uint64) bool {
	worker.jobStatusMutex.Lock()
	defer worker.jobStatusMutex.Unlock()
	if worker.jobStatusMap[id] {
		return true
	}
	worker.jobStatusMap[id] = true
	return false
}

func (worker *commonTaskWorker) jobDone(id uint64) {
	worker.jobStatusMutex.Lock()
	defer worker.jobStatusMutex.Unlock()
	delete(worker.jobStatusMap, id)
}

func (worker *commonTaskWorker) RunningMsInterval() time.Duration {
	return 100 * time.Millisecond
}

func (worker *commonTaskWorker) IsEnable() bool {
	return true
}