package ttl_task

import (
	"context"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/log"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/admin"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/job"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/scheduler"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/schema/ttl_task"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/types"
	pd "github.com/tikv/pd/client"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	batch "k8s.io/api/batch/v1"
	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	ctrl "sigs.k8s.io/controller-runtime/pkg/manager"
)

const (
	pollInterval                = time.Minute
	recycleDoneJobsInterval     = 30 * time.Second
	recycleOrphanJobsInterval   = 5 * time.Minute
	recycleJammedWorkerInterval = 1 * time.Minute
	maxWorkerLifetime           = 15 * time.Minute
	requeueDelay                = 15 * time.Minute

	// Records with an interval greater than maxNoTTLGap since the last execution are selected.
	maxNoTTLGap           = -15 * time.Minute
	updateMetricsInterval = 15 * time.Second

	ttlTaskName = "ttl"

	jobLoopWaitTime = 15 * time.Second
)

type handler struct {
	clusterCli               client.Client
	namespace                string
	wg                       sync.WaitGroup
	db                       *gorm.DB
	pdCli                    pd.Client
	jobBuilder               *job.Builder
	adminConfig              *admin.ConfigReconciler
	workerCertSecret         string
	jobLimit                 int
	taskRM                   *scheduler.Scheduler
	testRMkeyspaces          map[string]string
	enableTTLResourceManager bool
}

func (h *handler) NeedLeaderElection() bool {
	return true
}

func (h *handler) Start(ctx context.Context) error {
	log.Info("starting ttl task handler")

	// Init mysql table.
	err := h.db.WithContext(ctx).AutoMigrate(&ttl_task.TTLTableInfo{}, &ttl_task.Task{}, &ttl_task.TaskDone{})
	if err != nil {
		return err
	}

	initMetrics()

	h.wg.Add(5) //nolint:mnd // number of loops
	go h.pollLoop(ctx)
	go h.recycleDoneJobsLoop(ctx)
	go h.recycleOrphanJobsLoop(ctx)
	go h.recycleJammedWorkerLoop(ctx)
	go h.updateMetricsLoop(ctx)
	h.wg.Wait()

	return nil
}

func (h *handler) pollLoop(ctx context.Context) {
	defer h.wg.Done()

	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			err := h.poll(ctx)
			if err != nil {
				log.Error("failed to poll tasks", zap.Error(err))
				continue
			}
		}
	}
}

type PendingTask struct {
	KeyspaceName   string
	TidbPool       string
	TableID        int64
	LastTTLRunTime int64
	RegisteredAt   time.Time
}

func (h *handler) getEnableTTLKeyspaceFromTTLInfoTab(db *gorm.DB) ([]*ttl_task.TTLTableInfo, error) {
	var results []*ttl_task.TTLTableInfo

	findKs := `SELECT keyspace_name
    FROM ttl_table_infos
    WHERE is_enable = true AND paused = false
    GROUP BY keyspace_name
    HAVING COUNT(*) > 0`
	err := db.Raw(findKs).
		Scan(&results).Error
	return results, err
}

func (h *handler) selectTaskByKeyspace(db *gorm.DB, expectTime int64) ([]*PendingTask, error) {
	var pendingTasks []*PendingTask

	ttlTableInfo, err := h.getEnableTTLKeyspaceFromTTLInfoTab(db)
	if err != nil {
		return nil, err
	}

	// select pending tasks by keyspace name, only get 1 task from each keyspace now.
	findTaskByKeyspaceName := `SELECT keyspace_name,
        tidb_pool,
        table_id,
        last_ttl_run_time,
        registered_at
        FROM ttl_tasks
        WHERE state = ?
          and keyspace_name = ?
        and last_ttl_run_time < ? order by registered_at desc limit 1`

	// Find pending tasks by keyspace.
	for _, info := range ttlTableInfo {
		var keyspacePendingTasks []*PendingTask
		// Process each table info
		log.Info("Processing keyspace", zap.String("keyspace-name", info.KeyspaceName))
		err = db.Raw(findTaskByKeyspaceName,
			ttl_task.StatePending,
			info.KeyspaceName,
			expectTime,
		).
			Scan(&keyspacePendingTasks).Error
		if err != nil {
			log.Error("failed to query pending pendingTasks", zap.Error(err))
			continue
		}
		pendingTasks = append(pendingTasks, keyspacePendingTasks...)
	}
	// sort pendingTasks by registered_at
	sort.Slice(pendingTasks, func(i, j int) bool {
		return pendingTasks[i].RegisteredAt.Before(pendingTasks[j].RegisteredAt)
	})
	return pendingTasks, err
}

func (h *handler) poll(ctx context.Context) error {
	now := time.Now().Unix()
	expectTime := time.Now().Add(maxNoTTLGap).Unix()
	log.Info("ttl task start poll", zap.Int64("now", now))
	db := h.db.WithContext(ctx)

	// It matches of pkg/tidbworker/admin/admin.yaml
	if h.adminConfig != nil && h.adminConfig.IsPaused(ttlTaskName) {
		log.Info("skip poll tasks", zap.String("handler", ttlTaskName), zap.Int64("now", now))
		return nil
	}

	pendingTasks, err := h.selectTaskByKeyspace(db, expectTime)
	if err != nil {
		log.Error("failed to query pending pendingTasks", zap.Int64("now", now), zap.Error(err))
		return err
	}
	disabled := make(map[string]struct{})

	lastKeyspaceName := ""
	runningTask := 0
	for _, task := range pendingTasks {
		log.Info("loop task",
			zap.String("keyspace-name", task.KeyspaceName),
			zap.Int64("table-id", task.TableID),
			zap.Time("register-time", task.RegisteredAt),
			zap.Int64("now", now))
		if _, ok := disabled[task.KeyspaceName]; ok {
			continue
		}

		meta, err := h.pdCli.LoadKeyspace(ctx, task.KeyspaceName)
		if err != nil {
			log.Error("failed to load keyspace",
				zap.String("keyspace-name", task.KeyspaceName),
				zap.Int64("now", now),
				zap.Error(err))
			continue
		}

		// If the keyspace is not enabled, we need to mark the task as done.
		if meta.State != keyspacepb.KeyspaceState_ENABLED {
			log.Info("skip activating worker, keyspace is not enabled",
				zap.String("keyspace-name", task.KeyspaceName),
				zap.Int64("now", now))
			disabled[task.KeyspaceName] = struct{}{}
			continue
		}

		jobExist, err := job.CheckJobExists(h.clusterCli, h.namespace, task.KeyspaceName, string(job.TypeTTL))
		if err != nil {
			log.Error("failed to check if job exists",
				zap.String("keyspace-name", task.KeyspaceName),
				zap.Error(err))
			continue
		}
		if jobExist {
			log.Info("skip activating job, keyspace already have a running job",
				zap.String("keyspace-name", task.KeyspaceName))
			continue
		}

		if lastKeyspaceName != task.KeyspaceName {
			go func(keyspaceName string, tidbPool string, lastTTLRunTime int64, tableID int64, registeredAt time.Time) {
				log.Info("activating job",
					zap.String("keyspace-name", keyspaceName),
					zap.Int64("last-ttl-run-time", lastTTLRunTime),
					zap.Int64("table-id", tableID),
					zap.Int64("now", now))

				err := h.activateJob(ctx, keyspaceName, meta.Id, tidbPool, tableID, lastTTLRunTime)
				if err != nil {
					log.Error("failed to activate job",
						zap.String("keyspace-name", keyspaceName),
						zap.Int64("now", now),
						zap.Error(err))
					return
				}
				runnerCounter.Inc()
				taskQueueDuration.Observe(time.Since(registeredAt).Seconds())
			}(task.KeyspaceName, task.TidbPool, task.LastTTLRunTime, task.TableID, task.RegisteredAt)

			lastKeyspaceName = task.KeyspaceName

			runningTask++

			// Sleep jobLoopWaitTime, then continue to start the next job
			if runningTask%h.jobLimit == 0 {
				log.Info("The number of jobs started exceeds ttl-job-limit, "+
					"wait for a period of time to continue starting",
					zap.Int64("now", now))
				time.Sleep(jobLoopWaitTime)
			}
		}
	}

	err = h.updateDisableKeyspaceWorkerMeta(db, disabled)
	if err != nil {
		log.Error("failed to update disabled keyspaces' task meta",
			zap.Int64("now", now),
			zap.Error(err))
		return err
	}

	return nil
}

// pickTasks: get all deduplicate them, and then sort by registration time in ascending order.
func (h *handler) pickTasks(pendingTasks []*PendingTask) []*PendingTask {
	var tasks []*PendingTask
	// Deduplicate the data, this map only used to deduplicate.
	ksTasksMap := make(map[string]int)
	for _, task := range pendingTasks {
		if _, exists := ksTasksMap[task.KeyspaceName]; !exists {
			ksTasksMap[task.KeyspaceName] = 0
			task.TableID = 0
			tasks = append(tasks, task)
		}
	}

	// Sort tasks by `RegisteredAt` in asc order.
	sort.Slice(tasks, func(i, j int) bool {
		return tasks[i].RegisteredAt.Before(tasks[j].RegisteredAt)
	})
	return tasks
}

func (h *handler) isTestKeyspace(keyspaceName string) bool {
	log.Debug("check is test keyspace",
		zap.Any("test-keyspaces", h.testRMkeyspaces),
		zap.Any("temp-test-keyspaces", h.adminConfig.TestRMkeyspaces),
		zap.Bool("enable-ttl-resource-manager", h.enableTTLResourceManager))
	if _, exists := h.testRMkeyspaces[keyspaceName]; exists {
		log.Info("is test keyspace",
			zap.String("keyspace-name", keyspaceName))
		return true
	}

	if h.enableTTLResourceManager {
		log.Debug("enable ttl resource manager",
			zap.String("keyspace-name", keyspaceName))
		return true
	}

	// TestRMkeyspaces in adminConfig is a temporary list that will be cleared by a restart.
	if _, exists := h.adminConfig.TestRMkeyspaces[keyspaceName]; exists {
		log.Info("is temp test keyspace",
			zap.String("keyspace-name", keyspaceName))
		return true
	}

	return false
}

func (h *handler) updateDisableKeyspaceWorkerMeta(db *gorm.DB, disabled map[string]struct{}) error {
	// If keyspace is not enabled, the task state should be done.
	for k := range disabled {
		err := db.Transaction(func(tx *gorm.DB) error {
			log.Info("update disable keyspace meta, starting update old task to deleted",
				zap.String("keyspace-name", k))
			err := tx.Model(&ttl_task.Task{}).
				Where("keyspace_name = ?", k).
				Update("state", ttl_task.StateDeleted).Error
			if err != nil {
				log.Error("failed to update disabled keyspaces' task state",
					zap.String("keyspace-name", k), zap.Error(err))
				return err
			}

			table_info := &ttl_task.TTLTableInfo{KeyspaceName: k}
			err = tx.Delete(table_info).Error
			if err != nil {
				log.Error("failed to update disabled keyspaces' task state",
					zap.String("keyspace-name", k), zap.Error(err))
				return err
			}

			return nil
		})
		if err != nil {
			log.Error("failed to update disabled keyspaces' task meta",
				zap.String("keyspace-name", k), zap.Error(err))
			return err
		}
	}
	return nil
}

func (h *handler) recycleDoneJobsLoop(ctx context.Context) {
	defer h.wg.Done()

	ticker := time.NewTicker(recycleDoneJobsInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			err := h.recycleDoneJobs(ctx)
			if err != nil {
				log.Error("failed to recycle done tasks", zap.Error(err))
			}
		}
	}
}

// When job set done state, it just updates the state,but not recycle the k8s resource,
// it recycled by recycleDoneJobs finally.
func (h *handler) recycleDoneJobs(ctx context.Context) error {
	var (
		db                 = h.db.WithContext(ctx)
		doneOrDeletedTasks []ttl_task.Task
	)

	// Collect the runners that have finished their tasks.
	err := db.Model(&ttl_task.Task{}).
		Where("state = ?", ttl_task.StateDone).
		Or("state = ?", ttl_task.StateDeleted).
		Find(&doneOrDeletedTasks).Error
	if err != nil {
		log.Error("failed to list finished runner", zap.Error(err))
		return err
	}

	// deletedRunners is used to mark runners that have been terminated.
	deletedRunners := map[string]struct{}{"": {}}
	for _, doneOrDeletedTask := range doneOrDeletedTasks {
		// Delete target runners.
		// If the job is terminated, ignore this record.
		if _, ok := deletedRunners[doneOrDeletedTask.Runner]; !ok {
			j := &batch.Job{
				ObjectMeta: meta.ObjectMeta{
					Name:      doneOrDeletedTask.Runner,
					Namespace: h.namespace,
				},
			}
			log.Info("deleting k8s job", zap.String("job", j.Name))
			err := h.clusterCli.Delete(ctx, j, client.PropagationPolicy(meta.DeletePropagationBackground))
			if client.IgnoreNotFound(err) != nil {
				log.Error("failed to delete job", zap.String("job", j.Name), zap.Error(err))
				continue
			}
		}

		// Delete the task record, then insert the record into done table
		log.Info("deleting task record", zap.Any("done-or-delete-task", doneOrDeletedTask))

		if h.taskRM != nil && h.isTestKeyspace(doneOrDeletedTask.KeyspaceName) && doneOrDeletedTask.Runner != "" {
			log.Info("resource manager recycle resource", zap.Any("done-or-delete-task", doneOrDeletedTask))
			err = h.taskRM.RecycleResource(ctx,
				doneOrDeletedTask.KeyspaceName,
				doneOrDeletedTask.Runner)
			if err != nil {
				log.Error("resource manager recycle resource err",
					zap.Any("done-or-delete-task", doneOrDeletedTask), zap.Error(err))
				continue
			}
		}

		err = db.Transaction(func(tx *gorm.DB) error {
			err = tx.Delete(doneOrDeletedTask).Error
			if err != nil {
				log.Error("failed to delete task record", zap.Any("done-task", doneOrDeletedTask))
				return err
			}

			err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&ttl_task.TaskDone{
				KeyspaceName:   doneOrDeletedTask.KeyspaceName,
				TidbPool:       doneOrDeletedTask.TidbPool,
				TableID:        doneOrDeletedTask.TableID,
				LastTTLRunTime: doneOrDeletedTask.LastTTLRunTime,
				Runner:         doneOrDeletedTask.Runner,
				RunnerCreated:  doneOrDeletedTask.RunnerCreated,
				State:          doneOrDeletedTask.State,
				RegisteredAt:   doneOrDeletedTask.RegisteredAt,
			}).Error
			if err != nil {
				log.Error("failed to create done task record",
					zap.Any("done-task", doneOrDeletedTask), zap.Error(err))
				return err
			}

			if doneOrDeletedTask.State != ttl_task.StateDone {
				log.Info("skip add pending task", zap.Any("done-task", doneOrDeletedTask))
				return nil
			}

			// make the next new pending task.
			now := time.Now()
			pendingTask := &ttl_task.Task{
				KeyspaceName:   doneOrDeletedTask.KeyspaceName,
				TidbPool:       doneOrDeletedTask.TidbPool,
				TableID:        0,
				LastTTLRunTime: now.Unix(),
				RegisteredAt:   now,
			}

			log.Info("recycle done jobs, call add pending ttl task",
				zap.Any("pending-task", pendingTask))
			err = h.addPendingTask(tx, pendingTask)
			if err != nil {
				return err
			}
			return nil
		})
		if err != nil {
			continue
		}
		// Mark the job as terminated.
		deletedRunners[doneOrDeletedTask.Runner] = struct{}{}
		taskCompleteDuration.Observe(time.Since(doneOrDeletedTask.RegisteredAt).Seconds())
	}
	return nil
}

func (h *handler) addPendingTask(tx *gorm.DB, srcTask *ttl_task.Task) error {
	pendingTask := &ttl_task.Task{
		KeyspaceName:   srcTask.KeyspaceName,
		TidbPool:       srcTask.TidbPool,
		TableID:        srcTask.TableID,
		LastTTLRunTime: srcTask.LastTTLRunTime,
		State:          ttl_task.StatePending,
		RegisteredAt:   srcTask.RegisteredAt,
	}
	log.Info("insert into ttl_tasks, try to create a new pending task record",
		zap.Any("pending-task", pendingTask))
	err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&pendingTask).Error
	if err != nil {
		log.Error("failed to create pending toDeleteTask record",
			zap.Any("pendingTask", pendingTask), zap.Error(err))
		return err
	}

	var (
		toDeleteTasks []struct {
			KeyspaceName   string
			TidbPool       string
			TableID        int64
			LastTTLRunTime int64
			Runner         string
			state          string
		}
	)

	err = tx.Model(&ttl_task.Task{}).
		Where("keyspace_name = ?", srcTask.KeyspaceName).
		Where("table_id = ?", srcTask.TableID).
		Where("last_ttl_run_time < ?", srcTask.LastTTLRunTime).
		Where("state = ?", ttl_task.StatePending).
		Scan(&toDeleteTasks).
		Error
	if err != nil {
		log.Error("failed to list old toDeleteTask", zap.Error(err))
		return err
	}

	for _, toDeleteTask := range toDeleteTasks {
		log.Info("add pending task, update old task to deleted",
			zap.Any("toDeleteTask", toDeleteTask))
		err := tx.Model(&ttl_task.Task{}).
			Where("keyspace_name = ?", toDeleteTask.KeyspaceName).
			Where("table_id = ?", toDeleteTask.TableID).
			Where("last_ttl_run_time < ?", toDeleteTask.LastTTLRunTime).
			Where("state = ?", ttl_task.StatePending).
			Update("state", ttl_task.StateDeleted).
			Error

		if err != nil {
			return err
		}
	}

	return nil
}

func (h *handler) recycleOrphanJobsLoop(ctx context.Context) {
	defer h.wg.Done()

	ticker := time.NewTicker(recycleOrphanJobsInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			err := h.recycleOrphanJobs(ctx)
			if err != nil {
				log.Error("failed to recycle orphan jobs", zap.Error(err))
			}
		}
	}
}

func (h *handler) recycleOrphanJobs(ctx context.Context) error {
	var (
		db      = h.db.WithContext(ctx)
		results []struct {
			KeyspaceName string
			TableID      int64
		}
	)

	err := db.Model(&ttl_task.TTLTableInfo{}).
		Select("keyspace_name", "table_id").
		Scan(&results).Error
	if err != nil {
		log.Error("failed to list running tasks", zap.Error(err))
		return err
	}

	// Set 'done' state when keyspace is not enabled.
	for _, r := range results {
		m, err := h.pdCli.LoadKeyspace(ctx, r.KeyspaceName)
		if err != nil {
			log.Error("failed to load keyspace", zap.String("keyspace-name", r.KeyspaceName), zap.Error(err))
			continue
		}

		if m.State != keyspacepb.KeyspaceState_ENABLED && m.State != keyspacepb.KeyspaceState_DISABLED {
			log.Info("recycle orphan jobs, starting update old task to deleted",
				zap.String("keyspace-name", r.KeyspaceName), zap.Int64("table-id", r.TableID))

			err = db.Transaction(func(tx *gorm.DB) error {
				err := tx.Model(&ttl_task.Task{}).
					Where("keyspace_name = ?", r.KeyspaceName).
					Where("table_id = ?", r.TableID).
					Update("state", ttl_task.StateDeleted).
					Error
				if err != nil {
					log.Error("failed to delete orphan jobs", zap.Error(err))
					return err
				}

				toDeleteTTLTableInfo := &ttl_task.TTLTableInfo{
					KeyspaceName: r.KeyspaceName,
					TableID:      r.TableID,
				}

				err = tx.Model(&ttl_task.TTLTableInfo{}).
					Delete(&toDeleteTTLTableInfo).
					Error
				if err != nil {
					log.Error("failed to delete orphan jobs", zap.Error(err))
					return err
				}
				return nil
			})
			if err != nil {
				log.Error(
					"failed to recycle orphan jobs",
					zap.String("keyspace-name", r.KeyspaceName),
					zap.Int64("table-id", r.TableID),
					zap.Error(err))
				continue
			}
		}
	}

	return nil
}

func (h *handler) recycleJammedWorkerLoop(ctx context.Context) {
	defer h.wg.Done()
	ticker := time.NewTicker(recycleJammedWorkerInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			err := h.recycleJammedWorker(ctx)
			if err != nil {
				log.Error("failed to recycle jammed worker", zap.Error(err))
			}
		}
	}
}

func (h *handler) recycleJammedWorker(ctx context.Context) error {
	var (
		db = h.db.WithContext(ctx)
	)

	err := db.Transaction(func(tx *gorm.DB) error {
		var (
			runners []struct {
				KeyspaceName   string
				TidbPool       string
				TableID        int64
				LastTTLRunTime int64
				Runner         string
			}
			now       = time.Now()
			requeueTs = now.Add(requeueDelay)
		)

		err := tx.Model(&ttl_task.Task{}).
			Select("keyspace_name", "tidb_pool", "table_id", "last_ttl_run_time", "runner").
			Where("state = ?", ttl_task.StateRunning).
			Where("runner_created < ?", now.Add(-maxWorkerLifetime).Unix()).
			Scan(&runners).
			Error
		if err != nil {
			log.Error("failed to list jammed workers", zap.Error(err))
			return err
		}

		for _, runnerTask := range runners {
			log.Info("recycle running jammed jobs, starting update old task to deleted",
				zap.Any("task", runnerTask))
			err := tx.Model(&ttl_task.Task{}).
				Where("runner = ?", runnerTask.Runner).
				Update("state", ttl_task.StateDeleted).
				Error

			if err != nil {
				return err
			}

			pendingTask := &ttl_task.Task{
				KeyspaceName:   runnerTask.KeyspaceName,
				TidbPool:       runnerTask.TidbPool,
				TableID:        runnerTask.TableID,
				LastTTLRunTime: requeueTs.Unix(),
			}

			log.Info("recycle jammed worker, call add pending ttl task",
				zap.Any("pending-task", pendingTask))
			err = h.addPendingTask(tx, pendingTask)
			if err != nil {
				return err
			}
		}

		return nil
	})

	return err
}

func (h *handler) updateMetricsLoop(ctx context.Context) {
	defer h.wg.Done()

	ticker := time.NewTicker(updateMetricsInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			h.updateMetrics(ctx)
		}
	}
}

func (h *handler) updateMetrics(ctx context.Context) {
	var (
		db = h.db.WithContext(ctx)
	)

	type TaskCount struct {
		State ttl_task.State
		Count int
	}

	countMap := map[ttl_task.State]float64{
		ttl_task.StatePending: 0,
		ttl_task.StateRunning: 0,
		ttl_task.StateDone:    0,
		ttl_task.StateDeleted: 0,
	}

	//-------------------- count for non-pending tasks ---------------------

	var noPendingTaskCounts []TaskCount
	err := db.Model(&ttl_task.Task{}).
		Select("state, count(*) as count").
		Group("state").
		Where("state!= ?", ttl_task.StatePending).
		Scan(&noPendingTaskCounts).Error
	if err != nil {
		log.Error("failed to count tasks", zap.Error(err))
		return
	}

	for _, tc := range noPendingTaskCounts {
		countMap[tc.State] = float64(tc.Count)
	}

	//-------------------- count for non-pending tasks end ---------------------

	//-------------------- count for pending tasks ---------------------
	// Count tasks pending 3 hours ago
	var pendingTaskCount int64
	now := time.Now()
	threeHoursAgo := now.Add(-3 * time.Hour)
	err = db.Model(&ttl_task.Task{}).
		Where("registered_at <= ?", threeHoursAgo).
		Where("state = ?", ttl_task.StatePending).
		Count(&pendingTaskCount).
		Error
	if err != nil {
		log.Error("failed to count tasks", zap.Error(err))
		return
	}

	countMap[ttl_task.StatePending] = float64(pendingTaskCount)

	//-------------------- count for pending tasks end --------------------

	for state, count := range countMap {
		currentTaskCounter[state].Set(count)
	}
}

func NewHandler(
	namespace string,
	mgr ctrl.Manager,
	db *gorm.DB,
	pdCli pd.Client,
	taskRM *scheduler.Scheduler,
	ops ...option) types.Handler {
	h := &handler{
		clusterCli:      mgr.GetClient(),
		namespace:       namespace,
		db:              db,
		pdCli:           pdCli,
		jobBuilder:      job.NewBuilder(namespace, mgr.GetClient()),
		taskRM:          taskRM,
		testRMkeyspaces: make(map[string]string),
	}
	for _, op := range ops {
		op(h)
	}
	return h
}

type option func(h *handler)

func WithAdminConfig(adminConfig *admin.ConfigReconciler) option {
	return func(h *handler) {
		h.adminConfig = adminConfig
	}
}

func WithWorkerCertSecret(workerCertSecret string) option {
	return func(h *handler) {
		h.workerCertSecret = workerCertSecret
	}
}

func WithJobLimit(limit int) option {
	return func(h *handler) {
		h.jobLimit = limit
	}
}

func WithTestKeyspace(testRMKeyspace string) option {
	return func(h *handler) {
		h.testRMkeyspaces = make(map[string]string)
		keyspaces := strings.Split(testRMKeyspace, ",")
		for _, keyspace := range keyspaces {
			h.testRMkeyspaces[keyspace] = keyspace
			log.Info("with test keyspaces", zap.String("keyspace", keyspace))
		}
	}
}

func WithEnableResourceManager(enableTTLResourceManager bool) option {
	return func(h *handler) {
		h.enableTTLResourceManager = enableTTLResourceManager
	}
}