package report

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

	backoff "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"
	_import "github.com/pingcap-inc/clinic/pkg/task_manager/task/import"
	"github.com/pingcap-inc/clinic/pkg/trafficjobclient"
)

type ReportScannerService struct {
	kc          task.K8sClient
	globallimit *task.GlobalLimit
	queue       chan _import.Event
	tjStore     *trafficjobclient.TrafficComparisonReportStorage
	inFlightMu  sync.Mutex
	inFlight    map[string]struct{}
}

func NewReportScannerService(kc task.K8sClient, globallimit *task.GlobalLimit) *ReportScannerService {
	ss := ReportScannerService{
		kc:          kc,
		globallimit: globallimit,
		queue:       make(chan _import.Event, _import.QueueSize),
		tjStore:     trafficjobclient.NewTrafficComparisonReportStorage(context.Background()),
		inFlight:    make(map[string]struct{}),
	}
	log.Info("new scanner finish")
	return &ss
}

func (s *ReportScannerService) Start() {
	log.Info("start report scanner")
	for i := 0; i < _import.Concurrency; i++ {
		go s.start()
	}
	go s.scanLoop()
}

// scanLoop scan all pending job from db and send HealthReportJob to queue
func (s *ReportScannerService) scanLoop() {
	// TODO(shenjun): need to find a way to stop the ticker if scanner is closeable
	// check pending job periodically
	log.Info("start scan loop")
	ticker := time.NewTicker(_import.Wait)
	defer ticker.Stop()
	for ; true; <-ticker.C {
		s.scanPendingJobs()
	}
}

func (s *ReportScannerService) scanPendingJobs() {
	scanStart := time.Now()
	queueLenBefore := len(s.queue)
	queueCap := cap(s.queue)
	inFlight := s.inFlightSnapshot()
	usedSlots := inFlight.count
	if queueLenBefore > usedSlots {
		usedSlots = queueLenBefore
	}
	availableSlots := queueCap - usedSlots
	if availableSlots < 0 {
		availableSlots = 0
	}

	healthWaiting, comparisonWaiting, pendingReportWaiting, runningReportWaiting, err := s.countPendingJobs(inFlight)
	if err != nil {
		log.Error("count report jobs failed", zap.Error(err))
		return
	}

	remainingSlots := availableSlots
	queuedHealthReports := 0
	queuedComparisonReports := 0
	queuedPendingReports := 0
	queuedRunningReports := 0
	fetchedHealthReports := 0
	fetchedComparisonReports := 0
	fetchedPendingReports := 0
	fetchedRunningReports := 0

	if remainingSlots > 0 && healthWaiting > 0 {
		healthReports := make([]*db.HealthReport, 0, remainingSlots)
		healthOperation := func() error {
			healthReports = make([]*db.HealthReport, 0, remainingSlots)
			return model.FetchPendingOrRunningHealthReportJobsBatch(&healthReports, remainingSlots, inFlight.healthReportFileUUIDs)
		}
		if err := backoff.Retry(healthOperation, backoff.NewExponentialBackOff()); err != nil {
			log.Error("fetch health report jobs failed", zap.Error(err))
			return
		}
		fetchedHealthReports = len(healthReports)
		for _, h := range healthReports {
			if s.enqueue(NewHealthReportJob(h)) {
				queuedHealthReports++
				remainingSlots--
			}
			if remainingSlots <= 0 {
				break
			}
		}
	}

	if remainingSlots > 0 && comparisonWaiting > 0 {
		comparisonReports := make([]*db.ComparisonReport, 0, remainingSlots)
		comparisonOperation := func() error {
			comparisonReports = make([]*db.ComparisonReport, 0, remainingSlots)
			return model.FetchPendingOrRunningComparisonJobsBatch(&comparisonReports, remainingSlots, inFlight.comparisonReportIDs)
		}
		if err := backoff.Retry(comparisonOperation, backoff.NewExponentialBackOff()); err != nil {
			log.Error("fetch comparison report jobs failed", zap.Error(err))
			return
		}
		fetchedComparisonReports = len(comparisonReports)
		for _, cr := range comparisonReports {
			if s.enqueue(NewComparisonReportJob(cr)) {
				queuedComparisonReports++
				remainingSlots--
			}
			if remainingSlots <= 0 {
				break
			}
		}
	}

	if remainingSlots > 0 && pendingReportWaiting > 0 {
		reports, err := model.SelectReportByStatusBatch(db.StatusPending, remainingSlots, inFlight.reportUUIDs)
		if err != nil {
			log.Error("fetch pending report jobs failed", zap.Error(err))
			return
		}
		fetchedPendingReports = len(reports)
		for _, r := range reports {
			if s.enqueue(NewReportJob(r, false, s.tjStore)) {
				queuedPendingReports++
				remainingSlots--
			}
			if remainingSlots <= 0 {
				break
			}
		}
	}

	if remainingSlots > 0 && runningReportWaiting > 0 {
		reports, err := model.SelectReportByStatusBatch(db.StatusRunning, remainingSlots, inFlight.reportUUIDs)
		if err != nil {
			log.Error("fetch running report jobs failed", zap.Error(err))
			return
		}
		fetchedRunningReports = len(reports)
		for _, r := range reports {
			if s.enqueue(NewReportJob(r, true, s.tjStore)) {
				queuedRunningReports++
				remainingSlots--
			}
			if remainingSlots <= 0 {
				break
			}
		}
	}

	totalWaiting := healthWaiting + comparisonWaiting + pendingReportWaiting + runningReportWaiting
	totalQueued := queuedHealthReports + queuedComparisonReports + queuedPendingReports + queuedRunningReports
	deferredJobs := totalWaiting - int64(totalQueued)
	if deferredJobs < 0 {
		deferredJobs = 0
	}

	log.Info("report scan pending jobs summary",
		zap.Int64("db_waiting_jobs", totalWaiting),
		zap.Int64("health_report_waiting_jobs", healthWaiting),
		zap.Int64("comparison_report_waiting_jobs", comparisonWaiting),
		zap.Int64("pending_report_waiting_jobs", pendingReportWaiting),
		zap.Int64("running_report_waiting_jobs", runningReportWaiting),
		zap.Int("used_slots", usedSlots),
		zap.Int("available_slots", availableSlots),
		zap.Int("fetched_health_reports", fetchedHealthReports),
		zap.Int("fetched_comparison_reports", fetchedComparisonReports),
		zap.Int("fetched_pending_reports", fetchedPendingReports),
		zap.Int("fetched_running_reports", fetchedRunningReports),
		zap.Int("queued_health_reports", queuedHealthReports),
		zap.Int("queued_comparison_reports", queuedComparisonReports),
		zap.Int("queued_pending_reports", queuedPendingReports),
		zap.Int("queued_running_reports", queuedRunningReports),
		zap.Int64("deferred_jobs", deferredJobs),
		zap.Int("in_flight_jobs", s.inFlightCount()),
		zap.Int("in_flight_jobs_before", inFlight.count),
		zap.Int("queue_len_before", queueLenBefore),
		zap.Int("queue_len_after", len(s.queue)),
		zap.Int("queue_cap", queueCap),
		zap.Duration("scan_duration", time.Since(scanStart)),
	)
}

func (s *ReportScannerService) countPendingJobs(inFlight reportInFlightSnapshot) (int64, int64, int64, int64, error) {
	var healthWaiting, comparisonWaiting, pendingReportWaiting, runningReportWaiting int64
	operation := func() error {
		var err error
		healthWaiting, err = model.CountPendingOrRunningHealthReportJobs(inFlight.healthReportFileUUIDs)
		if err != nil {
			return err
		}
		comparisonWaiting, err = model.CountPendingOrRunningComparisonJobs(inFlight.comparisonReportIDs)
		if err != nil {
			return err
		}
		pendingReportWaiting, err = model.CountReportsByStatus(db.StatusPending, inFlight.reportUUIDs)
		if err != nil {
			return err
		}
		runningReportWaiting, err = model.CountReportsByStatus(db.StatusRunning, inFlight.reportUUIDs)
		return err
	}
	if err := backoff.Retry(operation, backoff.NewExponentialBackOff()); err != nil {
		return 0, 0, 0, 0, err
	}
	return healthWaiting, comparisonWaiting, pendingReportWaiting, runningReportWaiting, nil
}

func (s *ReportScannerService) enqueue(e _import.Event) bool {
	key := reportEventKey(e)
	if key == "" {
		log.Warn("report job has empty in-flight key", zap.String("event_type", reflect.TypeOf(e).String()))
		return false
	}
	if !s.markInFlight(key) {
		return false
	}
	select {
	case s.queue <- e:
		return true
	default:
		s.releaseInFlight(key)
		log.Warn("report queue filled before enqueue",
			zap.String("event_key", key),
			zap.Int("queue_len", len(s.queue)),
			zap.Int("queue_cap", cap(s.queue)),
		)
		return false
	}
}

type reportInFlightSnapshot struct {
	healthReportFileUUIDs []string
	comparisonReportIDs   []uint
	reportUUIDs           []string
	count                 int
}

func (s *ReportScannerService) inFlightSnapshot() reportInFlightSnapshot {
	s.inFlightMu.Lock()
	defer s.inFlightMu.Unlock()

	snapshot := reportInFlightSnapshot{
		healthReportFileUUIDs: make([]string, 0, len(s.inFlight)),
		comparisonReportIDs:   make([]uint, 0, len(s.inFlight)),
		reportUUIDs:           make([]string, 0, len(s.inFlight)),
		count:                 len(s.inFlight),
	}
	for key := range s.inFlight {
		switch {
		case strings.HasPrefix(key, "health_report:"):
			snapshot.healthReportFileUUIDs = append(snapshot.healthReportFileUUIDs, strings.TrimPrefix(key, "health_report:"))
		case strings.HasPrefix(key, "comparison_report:"):
			rawID := strings.TrimPrefix(key, "comparison_report:")
			if id, err := strconv.ParseUint(rawID, 10, 64); err == nil {
				snapshot.comparisonReportIDs = append(snapshot.comparisonReportIDs, uint(id))
			}
		case strings.HasPrefix(key, "report:"):
			snapshot.reportUUIDs = append(snapshot.reportUUIDs, strings.TrimPrefix(key, "report:"))
		}
	}
	return snapshot
}

func (s *ReportScannerService) inFlightCount() int {
	s.inFlightMu.Lock()
	defer s.inFlightMu.Unlock()
	return len(s.inFlight)
}

func (s *ReportScannerService) markInFlight(key string) bool {
	if key == "" {
		return false
	}
	s.inFlightMu.Lock()
	defer s.inFlightMu.Unlock()
	if s.inFlight == nil {
		s.inFlight = make(map[string]struct{})
	}
	if _, ok := s.inFlight[key]; ok {
		return false
	}
	s.inFlight[key] = struct{}{}
	return true
}

func (s *ReportScannerService) releaseInFlight(key string) {
	if key == "" {
		return
	}
	s.inFlightMu.Lock()
	defer s.inFlightMu.Unlock()
	delete(s.inFlight, key)
}

func reportEventKey(e _import.Event) string {
	switch job := e.(type) {
	case *HealthReportJob:
		if job == nil || job.h == nil {
			return ""
		}
		return "health_report:" + job.h.FileUUID
	case *ComparisonReportJob:
		if job == nil || job.cr == nil {
			return ""
		}
		return fmt.Sprintf("comparison_report:%d", job.cr.ID)
	case *ReportJob:
		if job == nil || job.h == nil {
			return ""
		}
		return "report:" + job.h.UUID
	default:
		return ""
	}
}

// start consume the s.queue and call `Run` on received event
func (s *ReportScannerService) start() {
	for e := range s.queue {
		key := reportEventKey(e)
		start := time.Now()
		queueLenAfterDequeue := len(s.queue)
		queueCap := cap(s.queue)
		func() {
			defer s.releaseInFlight(key)
			defer func() {
				if r := recover(); r != nil {
					log.Error("report job panicked",
						zap.Any("recover", r),
						zap.Stack("stack"),
						zap.String("event_key", key),
						zap.Int("queue_len", queueLenAfterDequeue),
						zap.Int("queue_cap", queueCap),
						zap.String("event_type", reflect.TypeOf(e).String()),
					)
				}
			}()
			if err := e.Run(s.kc); err != nil {
				log.Error("report job failed",
					zap.Error(err),
					zap.String("event_key", key),
					zap.Duration("duration", time.Since(start)),
					zap.Int("queue_len", queueLenAfterDequeue),
					zap.Int("queue_cap", queueCap),
					zap.String("event_type", reflect.TypeOf(e).String()),
				)
				return
			}
			cost := time.Since(start)
			if cost >= 10*time.Second {
				log.Warn("report job finished slowly",
					zap.String("event_key", key),
					zap.Duration("duration", cost),
					zap.Int("queue_len", queueLenAfterDequeue),
					zap.Int("queue_cap", queueCap),
					zap.String("event_type", reflect.TypeOf(e).String()),
				)
			}
		}()
	}
}