package _import

import (
	"context"
	"fmt"
	"reflect"
	"strings"
	"time"

	"github.com/cenkalti/backoff/v4"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/task_manager/task"
)

const (
	QueueSize   = 100 // s
	Concurrency = 10
	Wait        = 10 * time.Second
	CleanGap    = 1 * time.Minute
)

type Event interface {
	Run(task.K8sClient) error
}

type ScannerService struct {
	kc          task.K8sClient
	globallimit *task.GlobalLimit
	queue       chan Event
}

func NewScannerService(kc task.K8sClient, globallimit *task.GlobalLimit) *ScannerService {
	ss := ScannerService{
		kc:          kc,
		globallimit: globallimit,
		queue:       make(chan Event, QueueSize),
	}
	log.Info("new scanner finish")
	return &ss
}

func (s *ScannerService) Start() {
	for i := 0; i < Concurrency; i++ {
		go s.start()
	}
	go func() {
		for {
			time.Sleep(CleanGap)
			s.kc.CleanResource()
		}
	}()
	go s.scanLoop()
}

// scanLoop scan all pending job from db and send import event to queue
func (s *ScannerService) scanLoop() {
	// TODO(shenjun): need to find a way to stop the ticker if scanner is closeable
	// check pending job periodically
	ticker := time.NewTicker(time.Second * 10)
	defer ticker.Stop()
	for ; true; <-ticker.C {
		files := make([]*db.UploadFile, 0)
		operation := func() error {
			return model.FetchPendingJob(&files)
		}
		err := backoff.Retry(operation, backoff.NewExponentialBackOff())
		if err != nil {
			log.Error("create job failed", zap.Error(err))
		}
		for _, f := range files {
			// if the s.queue is full, just skip the job directly
			select {
			case s.queue <- newImporterEvent(f):
			default:
				log.Warn("queue is full, skip the job", zap.String("file_uuid", f.FileUUID))
			}
		}
	}
}

func (s *ScannerService) start() {
	for e := range s.queue {
		ctx, f := context.WithTimeout(context.Background(), Wait)
		available := s.globallimit.Touch(ctx)
		if !available {
			f()
			log.Error("no resource available")
			continue
		}
		if err := s.Execution(e); err != nil {
			log.Error("rebuild failed", zap.Error(err))
		}
		f()
	}
}

func (s *ScannerService) WaitRebuild(e Event) error {
	select {
	case s.queue <- e:
		return nil
	case <-time.After(Wait):
		return fmt.Errorf("wait queue full")
	}
}

func (s *ScannerService) Execution(e Event) error {
	return e.Run(s.kc)
}

type ImporterEvent struct {
	File *db.UploadFile
	Task []string
}

func (e *ImporterEvent) Run(kc task.K8sClient) error {
	// check if the status in db is unchanged (should be pending), if not, skip the job.
	// when we push two identical jobs to the queue
	// we need to drop the second job if the first job is finished
	f, err := model.QueryUploadFile(e.File.FileUUID)
	if err != nil {
		log.Error("query upload file failed", zap.Error(err))
		return err
	}
	var skip bool
	// e.Task is read from the rebuild tag value
	for _, v := range e.Task {
		if v == "rebuildMetric" && f.RebuildMetricStatus != string(db.StatusPending) {
			log.Warn("rebuildMetric abnormal",
				zap.String("file_uuid", e.File.FileUUID),
				zap.Strings("task", e.Task),
				zap.String("RebuildMetricStatus", f.RebuildMetricStatus),
			)
			skip = true
			break
		}
		if v == "rebuildLog" && f.RebuildLogStatus != string(db.StatusPending) {
			log.Warn("rebuildLog abnormal",
				zap.String("file_uuid", e.File.FileUUID),
				zap.Strings("task", e.Task),
				zap.String("RebuildLogStatus", f.RebuildLogStatus),
			)
			skip = true
			break
		}
		if v == "rebuildConfiguration" && f.RebuildConfigurationStatus != string(db.StatusPending) {
			log.Warn("rebuildConfiguration abnormal",
				zap.String("file_uuid", e.File.FileUUID),
				zap.Strings("task", e.Task),
				zap.String("RebuildConfigurationStatus", f.RebuildConfigurationStatus),
			)
			skip = true
			break
		}
	}
	if skip {
		log.Warn("skip the job", zap.String("file_uuid", e.File.FileUUID))
		return nil
	}
	// estimate the pvc request size
	pvSize := (int(e.File.FileLen/1024/1024/1024) + 1) * 20

	if e.File.DirSize > 0 {
		pvSize = e.File.DirSize/1024/1024/1024 + 10 + int(e.File.FileLen/1024/1024/1024)
	}
	if pvSize > 200 {
		pvSize = 200
	}
	log.Info("calculate pv size", zap.Int("size", pvSize))

	return kc.CreateJob(&task.CreateJobRequest{
		OrganizationID: fmt.Sprintf("%d", e.File.OrgID),
		PvcSize:        pvSize,
		Concurrency:    getRebuildConcurrency(e.File.OrgID),
		FileUUID:       e.File.FileUUID,
		Param: map[string]string{
			"-config":   "/etc/config/config.yaml",
			"-dataPath": "/data", "-fid": e.File.FileUUID,
			"-types":   strings.Join(e.Task, ","),
			"-cluster": e.File.ClusterID,
			"-org":     fmt.Sprintf("%d", e.File.OrgID),
		},
	})
}

func getRebuildConcurrency(orgID uint) int {
	// get rebuild concurrency
	quota, found, err := model.FindOrgQuotaByOrgID(orgID)
	if err != nil {
		log.Warn("find org quota failed", zap.Uint("orgID", orgID), zap.Error(err))
		return 1
	}

	if !found {
		return 1
	}

	return int(quota.RebuildConcurrency)
}

func newImporterEvent(f *db.UploadFile) *ImporterEvent {
	getPendingJob := func(u *db.UploadFile) []string {
		touchRebuild := make([]string, 0)
		rt := reflect.TypeOf(u).Elem()
		rv := reflect.ValueOf(u).Elem()

		for i := 0; i < rt.NumField(); i++ {
			v, ok := rt.Field(i).Tag.Lookup("rebuild")
			if !ok {
				continue
			}
			status, ok := rv.Field(i).Interface().(string)
			if !ok {
				continue
			}
			if status == string(db.StatusPending) {
				touchRebuild = append(touchRebuild, v)
			}
		}
		return touchRebuild
	}
	return &ImporterEvent{
		File: f,
		Task: getPendingJob(f),
	}
}