package healthcheck

import (
	"context"
	"encoding/json"
	"reflect"
	"sync"
	"time"

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

	"github.com/pingcap-inc/clinic/pkg/api/healthcheck"
	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/o11y"
	"github.com/pingcap-inc/clinic/pkg/o11y/urls"
	_import "github.com/pingcap-inc/clinic/pkg/task_manager/task/import"
)

const (
	// Concurrency is the number of concurrent workers
	Concurrency = 10
	// Wait is the interval between scans
	Wait = 30 * time.Second
)

// ScannerService scans for pending health check jobs and executes them
type ScannerService struct {
	cfg          *config.Config
	queue        chan _import.Event
	urlsResolver *urls.ClusterURLsResolver
	inFlightMu   sync.Mutex
	inFlight     map[string]struct{}
}

// NewScannerService creates a new health check scanner service
func NewScannerService(cfg *config.Config) *ScannerService {
	ss := &ScannerService{
		cfg:      cfg,
		queue:    make(chan _import.Event, _import.QueueSize),
		inFlight: make(map[string]struct{}),
	}

	// Initialize URL resolver for cluster URLs
	if cfg != nil && cfg.CentralO11yConfig.GSEndpoint != "" {
		o11ySharedOpts := []o11y.Option{
			o11y.TimeoutOption(40 * time.Second),
			o11y.MaxIdleConnOption(10),
		}
		o11yCli, err := o11y.New(
			cfg.CentralO11yConfig.GSEndpoint,
			cfg.CentralO11yConfig.DSEndpointTPL,
			cfg.CentralO11yConfig.ClientId,
			cfg.CentralO11yConfig.ClientSecret,
			cfg.CentralO11yConfig.Env,
			append(o11ySharedOpts, o11y.ByocDSEndpointOption(cfg.CentralO11yConfig.ByocDSEndpoint))...,
		)
		if err != nil {
			log.Error("failed to create o11y client for health check scanner", zap.Error(err))
		} else {
			ss.urlsResolver = urls.NewClusterURLsResolver(o11yCli)
			log.Info("url resolver initialized for health check scanner")
		}
	}

	log.Info("new health check scanner finish")
	return ss
}

// Start starts the scanner service
func (s *ScannerService) Start() {
	log.Info("start health check scanner")
	log.Info("health check scanner config",
		zap.Int("concurrency", Concurrency),
		zap.Duration("scan_interval", Wait),
		zap.Int("queue_capacity", cap(s.queue)),
		zap.Int("queue_size_const", _import.QueueSize),
	)

	// Start workers
	for i := 0; i < Concurrency; i++ {
		go s.startWorker(i)
	}

	// Start scan loop
	go s.scanLoop()
}

func (s *ScannerService) scanLoop() {
	log.Info("start health check scan loop")

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

	for ; true; <-ticker.C {
		s.scanPendingJobs()
		s.scanCronStrategies()
	}
}

func (s *ScannerService) scanPendingJobs() {
	scanStart := time.Now()
	queueLenBefore := len(s.queue)
	queueCap := cap(s.queue)
	excludeUUIDs, inFlightBefore := s.inFlightSnapshot()
	usedSlots := inFlightBefore
	if queueLenBefore > usedSlots {
		usedSlots = queueLenBefore
	}
	availableSlots := queueCap - usedSlots
	if availableSlots < 0 {
		availableSlots = 0
	}

	dbWaitingJobs, err := s.countPendingOrRunningJobs(excludeUUIDs)
	if err != nil {
		log.Error("count pending health check jobs failed", zap.Error(err))
		return
	}

	results := make([]*db.HealthCheckResult, 0)
	queuedCount := 0
	queuedPendingCount := 0
	queuedRunningCount := 0
	queuedPartialCount := 0

	if availableSlots > 0 && dbWaitingJobs > 0 {
		operation := func() error {
			results = make([]*db.HealthCheckResult, 0, availableSlots)
			return repository.FetchPendingOrRunningHealthCheckJobsBatch(&results, availableSlots, excludeUUIDs)
		}

		err = backoff.Retry(operation, backoff.NewExponentialBackOff())
		if err != nil {
			log.Error("fetch pending health check jobs failed", zap.Error(err))
			return
		}

		for _, result := range results {
			if result == nil || result.UUID == "" {
				log.Warn("fetched invalid health check job",
					zap.String("uuid", ""),
					zap.Int("queue_len", len(s.queue)),
					zap.Int("queue_cap", queueCap),
				)
				continue
			}
			if !s.markInFlight(result.UUID) {
				continue
			}
			select {
			case s.queue <- NewJob(s.cfg, result):
				queuedCount++
				switch db.StatusType(result.Status) {
				case db.StatusPending:
					queuedPendingCount++
				case db.StatusRunning:
					queuedRunningCount++
				case db.StatusPartial:
					queuedPartialCount++
				}
				log.Debug("queued health check job",
					zap.String("uuid", result.UUID),
					zap.String("cluster_id", result.ClusterID),
				)
			default:
				s.releaseInFlight(result.UUID)
				log.Warn("health check queue filled before enqueue",
					zap.String("uuid", result.UUID),
					zap.String("cluster_id", result.ClusterID),
					zap.String("status", result.Status),
					zap.Int("queue_len", len(s.queue)),
					zap.Int("queue_cap", queueCap),
				)
			}
		}
	}

	fetchedPendingCount, fetchedRunningCount, fetchedPartialCount := countHealthCheckJobStatuses(results)
	deferredJobs := dbWaitingJobs - int64(queuedCount)
	if deferredJobs < 0 {
		deferredJobs = 0
	}

	log.Info("health check scan pending jobs summary",
		zap.Int64("db_waiting_jobs", dbWaitingJobs),
		zap.Int("used_slots", usedSlots),
		zap.Int("available_slots", availableSlots),
		zap.Int("fetched_jobs", len(results)),
		zap.Int("fetched_pending_jobs", fetchedPendingCount),
		zap.Int("fetched_running_jobs", fetchedRunningCount),
		zap.Int("fetched_partial_jobs", fetchedPartialCount),
		zap.Int("queued_jobs", queuedCount),
		zap.Int("queued_pending_jobs", queuedPendingCount),
		zap.Int("queued_running_jobs", queuedRunningCount),
		zap.Int("queued_partial_jobs", queuedPartialCount),
		zap.Int64("deferred_jobs", deferredJobs),
		zap.Int("in_flight_jobs", s.inFlightCount()),
		zap.Int("in_flight_jobs_before", inFlightBefore),
		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 countHealthCheckJobStatuses(results []*db.HealthCheckResult) (pending, running, partial int) {
	for _, result := range results {
		if result == nil {
			continue
		}
		switch db.StatusType(result.Status) {
		case db.StatusPending:
			pending++
		case db.StatusRunning:
			running++
		case db.StatusPartial:
			partial++
		}
	}
	return pending, running, partial
}

func (s *ScannerService) countPendingOrRunningJobs(excludeUUIDs []string) (int64, error) {
	var count int64
	operation := func() error {
		var err error
		count, err = repository.CountPendingOrRunningHealthCheckJobs(excludeUUIDs)
		return err
	}
	if err := backoff.Retry(operation, backoff.NewExponentialBackOff()); err != nil {
		return 0, err
	}
	return count, nil
}

func (s *ScannerService) inFlightSnapshot() ([]string, int) {
	s.inFlightMu.Lock()
	defer s.inFlightMu.Unlock()

	uuids := make([]string, 0, len(s.inFlight))
	for uuid := range s.inFlight {
		uuids = append(uuids, uuid)
	}
	return uuids, len(s.inFlight)
}

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

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

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

func (s *ScannerService) scanCronStrategies() {
	// Fetch cron strategies that need to be triggered
	strategies, err := repository.ListPendingCronHealthCheckStrategies()
	if err != nil {
		log.Error("fetch pending cron health check strategies failed", zap.Error(err))
		return
	}

	now := time.Now()
	for _, strategy := range strategies {
		// Parse check items from strategy
		var items []healthcheck.CheckItem
		if strategy.Items != "" && strategy.Items != "null" {
			if err := json.Unmarshal([]byte(strategy.Items), &items); err != nil {
				log.Error("failed to parse strategy items",
					zap.Error(err),
					zap.String("strategy_uuid", strategy.UUID),
				)
				continue
			}
		}

		// Filter enabled items
		var enabledItems []healthcheck.CheckItem
		for _, item := range items {
			if item.Enabled {
				enabledItems = append(enabledItems, item)
			}
		}
		if len(enabledItems) == 0 {
			log.Info("strategy has no enabled items, skipping",
				zap.String("strategy_uuid", strategy.UUID),
			)
			continue
		}
		itemsToRunJSON, err := json.Marshal(enabledItems)
		if err != nil {
			log.Error("failed to marshal items json", zap.Error(err))
			continue
		}

		// Resolve cluster URLs if resolver is available
		var clusterURLsJSON string
		if s.urlsResolver != nil && s.urlsResolver.IsEnabled() {
			// Get cluster info for URL resolution
			cluster, err := model.GetAnyClusterWithOrg(strategy.ClusterID)
			if err != nil {
				log.Error("failed to get cluster for cron health check",
					zap.Error(err),
					zap.String("cluster_id", strategy.ClusterID),
					zap.String("strategy_uuid", strategy.UUID),
				)
				continue
			}

			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
			clusterURLs, err := s.urlsResolver.Resolve(ctx, cluster)
			cancel()

			if err != nil {
				log.Error("failed to resolve cluster URLs for cron health check",
					zap.Error(err),
					zap.String("cluster_id", strategy.ClusterID),
					zap.String("strategy_uuid", strategy.UUID),
				)
				continue
			}

			// Serialize cluster URLs to JSON
			urlsData := healthcheck.HealthCheckClusterUrls{
				ClusterMetricsUrl:   clusterURLs.ClusterMetricsURL,
				DataPlaneMetricsUrl: clusterURLs.DataPlaneMetricsURL,
				SlowQueryS3Url:      clusterURLs.SlowQueryS3URL,
			}
			urlsJSON, _ := json.Marshal(urlsData)
			clusterURLsJSON = string(urlsJSON)

			log.Info("resolved cluster URLs for cron health check",
				zap.String("cluster_id", strategy.ClusterID),
				zap.String("strategy_uuid", strategy.UUID),
				zap.String("cluster_metrics_url", urlsData.ClusterMetricsUrl),
			)
		}

		// Create single result for all enabled items
		result := &db.HealthCheckResult{
			UUID:         uuid.NewString(),
			StrategyUUID: strategy.UUID,
			ClusterID:    strategy.ClusterID,
			DbaasOrgID:   strategy.DbaasOrgID,
			ItemsToRun:   string(itemsToRunJSON),
			Status:       string(db.StatusPending),
			ClusterURLs:  clusterURLsJSON,
		}

		// Parse strategy's time range to absolute times for the result
		// Prefer _Expr fields (new), fallback to legacy fields for backward compatibility
		timeRangeAStart := getTimeString(strategy.TimeRangeAStartExpr, strategy.TimeRangeAStart)
		timeRangeAEnd := getTimeString(strategy.TimeRangeAEndExpr, strategy.TimeRangeAEnd)
		timeRangeBStart := getTimeString(strategy.TimeRangeBStartExpr, strategy.TimeRangeBStart)
		timeRangeBEnd := getTimeString(strategy.TimeRangeBEndExpr, strategy.TimeRangeBEnd)

		// Range A may be an "HH:MM" inspection window resolved against the trigger
		// time (now) in the cron's timezone, or a legacy expression. ResolveRangeA
		// encapsulates both paths (and is unit-tested directly).
		result.TimeRangeAStart, result.TimeRangeAEnd = healthcheck.ResolveRangeA(
			timeRangeAStart, timeRangeAEnd, strategy.ScheduleExpr, now,
		)
		if timeRangeBStart != "" {
			t, err := healthcheck.ParseTimeString(timeRangeBStart)
			if err == nil {
				result.TimeRangeBStart = t
			}
		}
		if timeRangeBEnd != "" {
			t, err := healthcheck.ParseTimeString(timeRangeBEnd)
			if err == nil {
				result.TimeRangeBEnd = t
			}
		}

		if err := repository.InsertHealthCheckResult(result); err != nil {
			log.Error("insert health check result for cron job failed",
				zap.Error(err),
				zap.String("strategy_uuid", strategy.UUID),
			)
			continue
		}

		log.Info("created health check result for cron job",
			zap.String("strategy_uuid", strategy.UUID),
			zap.String("result_uuid", result.UUID),
			zap.Int("item_count", len(enabledItems)),
		)

		// Update strategy's last run time and next run time
		nextRunAt := calculateNextRunTimeFromNow(strategy.ScheduleExpr)
		if nextRunAt == nil {
			// Cron parsing failed, pause the strategy to prevent infinite loop
			log.Error("cron expression parsing failed, pausing strategy",
				zap.String("strategy_uuid", strategy.UUID),
				zap.String("schedule_expr", strategy.ScheduleExpr),
			)
			if err := repository.UpdateHealthCheckStrategyStatus(strategy.UUID, "paused", "invalid cron expression: "+strategy.ScheduleExpr); err != nil {
				log.Error("failed to pause strategy with invalid cron expression",
					zap.Error(err),
					zap.String("strategy_uuid", strategy.UUID),
				)
			}
			continue
		}
		if err := repository.UpdateHealthCheckStrategyLastRun(strategy.UUID, &now, nextRunAt); err != nil {
			log.Error("update health check strategy last run time failed",
				zap.Error(err),
				zap.String("strategy_uuid", strategy.UUID),
			)
		}
	}
}

func (s *ScannerService) startWorker(workerID int) {
	log.Info("health check worker started",
		zap.Int("worker_id", workerID),
		zap.Int("queue_capacity", cap(s.queue)),
	)
	for e := range s.queue {
		start := time.Now()
		queueLenAfterDequeue := len(s.queue)
		queueCap := cap(s.queue)
		var jobUUID, clusterID string
		if j, ok := e.(*Job); ok && j != nil && j.result != nil {
			jobUUID = j.result.UUID
			clusterID = j.result.ClusterID
		}

		func() {
			defer s.releaseInFlight(jobUUID)
			defer func() {
				if r := recover(); r != nil {
					log.Error("health check job panicked",
						zap.Any("recover", r),
						zap.Stack("stack"),
						zap.Int("worker_id", workerID),
						zap.String("uuid", jobUUID),
						zap.String("cluster_id", clusterID),
						zap.Int("queue_len", queueLenAfterDequeue),
						zap.Int("queue_cap", queueCap),
						zap.String("event_type", reflect.TypeOf(e).String()),
					)
				}
			}()
			if err := e.Run(nil); err != nil {
				log.Error("health check job failed",
					zap.Error(err),
					zap.Int("worker_id", workerID),
					zap.String("uuid", jobUUID),
					zap.String("cluster_id", clusterID),
					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("health check job finished slowly",
					zap.Int("worker_id", workerID),
					zap.String("uuid", jobUUID),
					zap.String("cluster_id", clusterID),
					zap.Duration("duration", cost),
					zap.Int("queue_len", queueLenAfterDequeue),
					zap.Int("queue_cap", queueCap),
					zap.String("event_type", reflect.TypeOf(e).String()),
				)
			}
		}()
	}
}

// calculateNextRunTimeFromNow calculates the next run time from now based on the cron expression
// This is a convenience wrapper that calls calculateNextRunTime with current time
func calculateNextRunTimeFromNow(cronExpr string) *time.Time {
	return calculateNextRunTime(cronExpr, time.Now())
}

// getTimeString returns the time string, preferring the expression field
// Falls back to legacy time field formatted as RFC3339 if expression is empty
func getTimeString(expr string, legacy *time.Time) string {
	if expr != "" {
		return expr
	}
	if legacy != nil {
		return legacy.Format(time.RFC3339)
	}
	return ""
}