package diskcheck

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/aws/aws-sdk-go/service/sts"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/api/healthcheck"
	"github.com/pingcap-inc/clinic/pkg/o11y"
)

// ============================================
// Parameter Configuration
// ============================================

// DiskCheckParams represents configurable parameters for disk check
type DiskCheckParams struct {
	// AWS EBS disk latency thresholds (in seconds)
	IO2Threshold float64 `json:"io2_threshold,omitempty"` // default: 0.005 (5ms)
	GP3Threshold float64 `json:"gp3_threshold,omitempty"` // default: 0.01 (10ms)

	// GCP PD disk latency thresholds (in seconds)
	PDSSDThreshold      float64 `json:"pd_ssd_threshold,omitempty"`      // default: 0.008 (8ms)
	PDBalancedThreshold float64 `json:"pd_balanced_threshold,omitempty"` // default: 0.010 (10ms)
	PDExtremeThreshold  float64 `json:"pd_extreme_threshold,omitempty"`  // default: 0.005 (5ms)
	PDStandardThreshold float64 `json:"pd_standard_threshold,omitempty"` // default: 0.015 (15ms)

	// Time windows for aggregation and queries
	WindowMergeInterval   int `json:"window_merge_interval_minutes,omitempty"` // default: 5
	BaselineDurationHours int `json:"baseline_duration_hours,omitempty"`       // default: 2
	PeakBufferMinutes     int `json:"peak_buffer_minutes,omitempty"`           // default: 3
	RaftBufferMinutes     int `json:"raft_buffer_minutes,omitempty"`           // default: 5
	EBSBufferMinutes      int `json:"ebs_buffer_minutes,omitempty"`            // default: 10
	GCPBufferMinutes      int `json:"gcp_buffer_minutes,omitempty"`            // default: 10
	TiFlashMinDurationMin int `json:"tiflash_min_duration_minutes,omitempty"`  // default: 5

	// Impact evaluation thresholds
	ImpactRatioThreshold float64 `json:"impact_ratio_threshold,omitempty"` // default: 1.5
	CriticalLevelRatio   float64 `json:"critical_level_ratio,omitempty"`   // default: 3.0
	MajorLevelRatio      float64 `json:"major_level_ratio,omitempty"`      // default: 2.0
	WarningLevelRatio    float64 `json:"warning_level_ratio,omitempty"`    // default: 1.5
}

// DefaultParams returns default parameters
func DefaultParams() DiskCheckParams {
	return DiskCheckParams{
		// AWS EBS thresholds
		IO2Threshold: 0.005, // 5ms
		GP3Threshold: 0.01,  // 10ms
		// GCP PD thresholds
		PDSSDThreshold:      0.008, // 8ms
		PDBalancedThreshold: 0.010, // 10ms
		PDExtremeThreshold:  0.005, // 5ms
		PDStandardThreshold: 0.015, // 15ms
		// Time windows
		WindowMergeInterval:   5,
		BaselineDurationHours: 2,
		PeakBufferMinutes:     3,
		RaftBufferMinutes:     5,
		EBSBufferMinutes:      10,
		GCPBufferMinutes:      10,
		TiFlashMinDurationMin: 5,
		// Impact thresholds
		ImpactRatioThreshold: 1.5,
		CriticalLevelRatio:   3.0,
		MajorLevelRatio:      2.0,
		WarningLevelRatio:    1.5,
	}
}

// ParseParams parses parameters from request, using defaults for missing values
func ParseParams(params json.RawMessage) DiskCheckParams {
	p := DefaultParams()
	if len(params) == 0 {
		return p
	}

	if err := json.Unmarshal(params, &p); err != nil {
		log.Warn("failed to parse disk check params, using defaults", zap.Error(err))
		return DefaultParams()
	}

	// Validate and set defaults for zero values
	if p.IO2Threshold == 0 {
		p.IO2Threshold = DefaultParams().IO2Threshold
	}
	if p.GP3Threshold == 0 {
		p.GP3Threshold = DefaultParams().GP3Threshold
	}
	// GCP PD thresholds
	if p.PDSSDThreshold == 0 {
		p.PDSSDThreshold = DefaultParams().PDSSDThreshold
	}
	if p.PDBalancedThreshold == 0 {
		p.PDBalancedThreshold = DefaultParams().PDBalancedThreshold
	}
	if p.PDExtremeThreshold == 0 {
		p.PDExtremeThreshold = DefaultParams().PDExtremeThreshold
	}
	if p.PDStandardThreshold == 0 {
		p.PDStandardThreshold = DefaultParams().PDStandardThreshold
	}
	if p.WindowMergeInterval == 0 {
		p.WindowMergeInterval = DefaultParams().WindowMergeInterval
	}
	if p.BaselineDurationHours == 0 {
		p.BaselineDurationHours = DefaultParams().BaselineDurationHours
	}
	if p.PeakBufferMinutes == 0 {
		p.PeakBufferMinutes = DefaultParams().PeakBufferMinutes
	}
	if p.RaftBufferMinutes == 0 {
		p.RaftBufferMinutes = DefaultParams().RaftBufferMinutes
	}
	if p.EBSBufferMinutes == 0 {
		p.EBSBufferMinutes = DefaultParams().EBSBufferMinutes
	}
	if p.GCPBufferMinutes == 0 {
		p.GCPBufferMinutes = DefaultParams().GCPBufferMinutes
	}
	if p.TiFlashMinDurationMin == 0 {
		p.TiFlashMinDurationMin = DefaultParams().TiFlashMinDurationMin
	}
	if p.ImpactRatioThreshold == 0 {
		p.ImpactRatioThreshold = DefaultParams().ImpactRatioThreshold
	}
	if p.CriticalLevelRatio == 0 {
		p.CriticalLevelRatio = DefaultParams().CriticalLevelRatio
	}
	if p.MajorLevelRatio == 0 {
		p.MajorLevelRatio = DefaultParams().MajorLevelRatio
	}
	if p.WarningLevelRatio == 0 {
		p.WarningLevelRatio = DefaultParams().WarningLevelRatio
	}

	return p
}

// ============================================
// Raw Detection Data
// ============================================

// DiskAnomalyWindow represents a disk anomaly time window
type DiskAnomalyWindow struct {
	StartTime    time.Time         `json:"start_time"`
	EndTime      time.Time         `json:"end_time"`
	MetricName   string            `json:"metric_name"`
	InstanceName string            `json:"instance_name"`
	Component    string            `json:"component"`
	MaxValue     float64           `json:"max_value"`
	DiskType     string            `json:"disk_type"`
	Labels       map[string]string `json:"labels,omitempty"`
}

// EBSMetrics represents EBS throughput exceeded metrics
type EBSMetrics struct {
	ThroughputExceeded float64 `json:"throughput_exceeded"`
	IOPSExceeded       float64 `json:"iops_exceeded"`
	Throughput         float64 `json:"throughput"`
}

// GCPDiskMetrics represents GCP PD performance metrics
type GCPDiskMetrics struct {
	IOPSExceeded       float64 `json:"iops_exceeded"`       // IOPS quota exceeded
	ThroughputExceeded float64 `json:"throughput_exceeded"` // Throughput quota exceeded
	ReadIOPS           float64 `json:"read_iops"`           // Read IOPS
	WriteIOPS          float64 `json:"write_iops"`          // Write IOPS
	ReadThroughput     float64 `json:"read_throughput"`     // Read throughput
	WriteThroughput    float64 `json:"write_throughput"`    // Write throughput
}

// TiDBImpact represents TiDB performance impact
type TiDBImpact struct {
	Baseline float64 `json:"baseline"`
	Peak     float64 `json:"peak"`
	Ratio    float64 `json:"ratio"`
}

// TiFlashImpact represents TiFlash performance impact
type TiFlashImpact struct {
	Baseline                   float64 `json:"baseline"`
	Peak                       float64 `json:"peak"`
	Ratio                      float64 `json:"ratio"`
	RaftWaitIndexDuration      float64 `json:"raft_wait_index_duration"`
	RaftBatchReadIndexDuration float64 `json:"raft_batch_read_index_duration"`
}

// RawDiskAnomalyEvent represents a complete raw disk anomaly event (before aggregation)
type RawDiskAnomalyEvent struct {
	StartTime       time.Time      `json:"start_time"`
	EndTime         time.Time      `json:"end_time"`
	InstanceName    string         `json:"instance_name"`
	Component       string         `json:"component"`
	DiskType        string         `json:"disk_type"`
	MaxReadLatency  float64        `json:"max_read_latency"`
	MaxWriteLatency float64        `json:"max_write_latency"`
	EBS             *EBSMetrics    `json:"ebs,omitempty"`
	TiDB            *TiDBImpact    `json:"tidb,omitempty"`
	TiFlash         *TiFlashImpact `json:"tiflash,omitempty"`
	AnomalyLevel    string         `json:"anomaly_level"`
}

// ============================================
// Report Format
// ============================================

// DiskHealthReport represents the final health check report in PDF-like format
type DiskHealthReport struct {
	ClusterID      string          `json:"cluster_id"`
	ReportID       string          `json:"report_id"`
	CheckTime      time.Time       `json:"check_time"`
	TimeRangeStart string          `json:"time_range_start"`
	TimeRangeEnd   string          `json:"time_range_end"`
	Params         DiskCheckParams `json:"params"` // Record used parameters
	Summary        SummaryStats    `json:"summary"`
	CaseTracking   CaseTracking    `json:"case_tracking"`
}

// SummaryStats represents the summary section of the report
type SummaryStats struct {
	TotalDisks DiskCount   `json:"total_disks"`
	Statistics []DiskStats `json:"statistics"`
}

// DiskCount represents disk count by type and component
type DiskCount struct {
	// AWS EBS
	IO2 int `json:"io2"`
	GP3 int `json:"gp3"`
	// GCP PD
	PDSSD      int `json:"pd_ssd"`
	PDBalanced int `json:"pd_balanced"`
	PDExtreme  int `json:"pd_extreme"`
	PDStandard int `json:"pd_standard"`
	// Components
	TiKV    int `json:"tikv"`
	TiFlash int `json:"tiflash"`
}

// DiskStats represents statistics for each disk type and impact type combination
type DiskStats struct {
	DiskType           string `json:"disk_type"`            // io2, gp3, pd_ssd, pd_balanced, pd_extreme, pd_standard
	ImpactType         string `json:"impact_type"`          // Impacted, Non-Impacted
	JitterEventsCount  int    `json:"jitter_events_count"`  // Unknown cause
	TiKVIssuesCount    int    `json:"tikv_issues_count"`    // EBS/GCP exceeded + tikv
	TiFlashIssuesCount int    `json:"tiflash_issues_count"` // EBS/GCP exceeded + tiflash
}

// CaseTracking represents the case tracking section
type CaseTracking struct {
	Impacted    ImpactedCases    `json:"impacted"`
	NonImpacted NonImpactedCases `json:"non_impacted"`
}

// ImpactedCases represents impacted cases (with business impact)
type ImpactedCases struct {
	TiDBIssues   []EventRecord `json:"tidb_issues"`
	JitterEvents []EventRecord `json:"jitter_events"`
}

// NonImpactedCases represents non-impacted cases (no business impact)
type NonImpactedCases struct {
	TiDBIssues   []EventRecord `json:"tidb_issues"`
	JitterEvents []EventRecord `json:"jitter_events"`
}

// EventRecord represents a single event record in the report (aggregated by time window)
type EventRecord struct {
	StartTime              time.Time `json:"start_time"`
	EndTime                time.Time `json:"end_time"`
	ClusterID              string    `json:"cluster_id"`
	DiskType               string    `json:"disk_type"`     // io2, gp3
	Duration               string    `json:"duration"`      // "30s", "5m30s"
	JitterLevel            string    `json:"jitter_level"`  // critical, major, warning, minor
	ImpactScope            string    `json:"impact_scope"`  // "tikv 1", "tiflash 6", "tikv 1, tiflash 2"
	RootCause              string    `json:"root_cause"`    // "EBS iops exceeded", "EBS throughput exceeded", "Unknown cause"
	JitterImpact           string    `json:"jitter_impact"` // Description of business impact
	DiskMonitoringLink     string    `json:"disk_monitoring_link,omitempty"`
	BusinessMonitoringLink string    `json:"business_monitoring_link,omitempty"`
}

// EventClassification represents the classification of an event
type EventClassification struct {
	IsTiDBIssue bool   // true = TiDB Issue (EBS exceeded), false = Jitter Event (Unknown cause)
	IsImpacted  bool   // true = has business impact
	IsTiKV      bool   // true = tikv, false = tiflash
	DiskType    string // io2, gp3
}

// Prometheus query result structure
type QueryResult struct {
	Status string `json:"status"`
	Data   struct {
		ResultType string `json:"resultType"`
		Result     []struct {
			Metric map[string]string `json:"metric"`
			Values [][]interface{}   `json:"values"`
		} `json:"result"`
	} `json:"data"`
}

// AsyncRun performs async disk check for a specific cluster and generates a report
func AsyncRun(req *healthcheck.HealthCheckJobRequest) error {
	// Parse parameters with defaults
	params := ParseParams(req.Params)

	log.Info("start disk check",
		zap.String("cluster_id", req.ClusterId),
		zap.String("report_id", req.ReportId),
		zap.String("deploy_type", req.DeployType),
		zap.String("time_range_start", req.TimeRange.Start),
		zap.String("time_range_end", req.TimeRange.End),
		zap.Float64("io2_threshold", params.IO2Threshold),
		zap.Float64("gp3_threshold", params.GP3Threshold),
		zap.Int("window_merge_interval", params.WindowMergeInterval),
		zap.Int("baseline_duration_hours", params.BaselineDurationHours),
	)

	// Parse time range
	startTime, endTime, err := parseTimeRange(req.TimeRange)
	if err != nil {
		log.Error("parse time range failed", zap.Error(err))
		return fmt.Errorf("parse time range failed: %w", err)
	}

	// DataPlaneMetricsUrl for node_xxx metrics (disk latency)
	// ClusterMetricsUrl for tidb_xxx, tiflash_xxx, tikv_xxx metrics
	if req.ClusterUrls.DataPlaneMetricsUrl == "" {
		return fmt.Errorf("data plane metrics URL is empty")
	}
	if req.ClusterUrls.ClusterMetricsUrl == "" {
		return fmt.Errorf("cluster metrics URL is empty")
	}

	// Get PromQL based on deploy type
	diskReadLatencyPromQL := getDiskReadLatencyPromQL(req.DeployType, req.ClusterId)
	diskWriteLatencyPromQL := getDiskWriteLatencyPromQL(req.DeployType, req.ClusterId)

	// Detect anomaly windows for read latency (from ClusterMetricsUrl - node_xxx metrics)
	readWindows, err := detectAnomalyWindows(req.ClusterUrls.ClusterMetricsUrl, "disk_read_latency", diskReadLatencyPromQL, startTime, endTime, params)
	if err != nil {
		log.Error("detect read latency anomaly windows failed", zap.Error(err))
		// Continue processing, don't return error
	}

	// Detect anomaly windows for write latency (from ClusterMetricsUrl - node_xxx metrics)
	writeWindows, err := detectAnomalyWindows(req.ClusterUrls.ClusterMetricsUrl, "disk_write_latency", diskWriteLatencyPromQL, startTime, endTime, params)
	if err != nil {
		log.Error("detect write latency anomaly windows failed", zap.Error(err))
		// Continue processing, don't return error
	}

	// Merge and deduplicate windows
	allWindows := mergeWindows(readWindows, writeWindows, params)

	// Build raw anomaly events with additional metrics
	// Pass both URLs: DataPlaneMetricsUrl for EBS/node metrics, ClusterMetricsUrl for TiDB/TiFlash metrics
	rawEvents := buildRawAnomalyEvents(req.ClusterUrls.DataPlaneMetricsUrl, req.ClusterUrls.ClusterMetricsUrl, req.DeployType, req.ClusterId, req.PhysicalClusterId, allWindows, params)

	// Generate the final report
	report := generateReport(req, rawEvents, startTime, endTime, params)

	// Upload result to S3
	s3Key := fmt.Sprintf("healthcheck-report/%s_%s_disk_check.json", req.ClusterId, req.ReportId)
	if err := uploadToS3(s3Key, report); err != nil {
		log.Error("upload result to s3 failed", zap.Error(err), zap.String("s3_key", s3Key))
		return fmt.Errorf("upload result to s3 failed: %w", err)
	}

	log.Info("disk check completed",
		zap.String("cluster_id", req.ClusterId),
		zap.String("report_id", req.ReportId),
		zap.Int("total_events", len(rawEvents)),
		zap.String("s3_key", s3Key),
	)
	return nil
}

// generateReport generates the final report from raw events
func generateReport(req *healthcheck.HealthCheckJobRequest, rawEvents []RawDiskAnomalyEvent, startTime, endTime time.Time, params DiskCheckParams) *DiskHealthReport {
	// Aggregate events by time window
	aggregatedEvents := aggregateEventsByTimeWindow(req.ClusterId, rawEvents, params)

	// Classify and categorize events
	var tidbIssuesImpacted, tidbIssuesNonImpacted []EventRecord
	var jitterEventsImpacted, jitterEventsNonImpacted []EventRecord

	for _, event := range aggregatedEvents {
		classification := classifyEvent(event)

		if classification.IsTiDBIssue {
			if classification.IsImpacted {
				tidbIssuesImpacted = append(tidbIssuesImpacted, event)
			} else {
				tidbIssuesNonImpacted = append(tidbIssuesNonImpacted, event)
			}
		} else {
			if classification.IsImpacted {
				jitterEventsImpacted = append(jitterEventsImpacted, event)
			} else {
				jitterEventsNonImpacted = append(jitterEventsNonImpacted, event)
			}
		}
	}

	// Build statistics
	stats := buildStatistics(aggregatedEvents)

	// Count disks by type and component
	diskCount := countDisks(rawEvents)

	return &DiskHealthReport{
		ClusterID:      req.ClusterId,
		ReportID:       req.ReportId,
		CheckTime:      time.Now(),
		TimeRangeStart: req.TimeRange.Start,
		TimeRangeEnd:   req.TimeRange.End,
		Params:         params, // Record used parameters
		Summary: SummaryStats{
			TotalDisks: diskCount,
			Statistics: stats,
		},
		CaseTracking: CaseTracking{
			Impacted: ImpactedCases{
				TiDBIssues:   tidbIssuesImpacted,
				JitterEvents: jitterEventsImpacted,
			},
			NonImpacted: NonImpactedCases{
				TiDBIssues:   tidbIssuesNonImpacted,
				JitterEvents: jitterEventsNonImpacted,
			},
		},
	}
}

// aggregateEventsByTimeWindow aggregates raw events by time window
func aggregateEventsByTimeWindow(clusterID string, rawEvents []RawDiskAnomalyEvent, params DiskCheckParams) []EventRecord {
	if len(rawEvents) == 0 {
		return nil
	}

	// Sort events by start time
	sortedEvents := make([]RawDiskAnomalyEvent, len(rawEvents))
	copy(sortedEvents, rawEvents)
	for i := 0; i < len(sortedEvents)-1; i++ {
		for j := i + 1; j < len(sortedEvents); j++ {
			if sortedEvents[i].StartTime.After(sortedEvents[j].StartTime) {
				sortedEvents[i], sortedEvents[j] = sortedEvents[j], sortedEvents[i]
			}
		}
	}

	var aggregated []EventRecord
	var currentGroup []RawDiskAnomalyEvent
	var currentEnd time.Time
	mergeInterval := time.Duration(params.WindowMergeInterval) * time.Minute

	for _, event := range sortedEvents {
		if len(currentGroup) == 0 {
			currentGroup = append(currentGroup, event)
			currentEnd = event.EndTime
			continue
		}

		// Check if within merge interval of current group
		if event.StartTime.Sub(currentEnd) <= mergeInterval {
			currentGroup = append(currentGroup, event)
			if event.EndTime.After(currentEnd) {
				currentEnd = event.EndTime
			}
		} else {
			// Create aggregated event from current group
			aggregated = append(aggregated, createAggregatedEvent(clusterID, currentGroup, params))
			currentGroup = []RawDiskAnomalyEvent{event}
			currentEnd = event.EndTime
		}
	}

	// Don't forget the last group
	if len(currentGroup) > 0 {
		aggregated = append(aggregated, createAggregatedEvent(clusterID, currentGroup, params))
	}

	return aggregated
}

// createAggregatedEvent creates a single EventRecord from a group of raw events
func createAggregatedEvent(clusterID string, events []RawDiskAnomalyEvent, params DiskCheckParams) EventRecord {
	if len(events) == 0 {
		return EventRecord{}
	}

	// Find time range
	startTime := events[0].StartTime
	endTime := events[0].EndTime
	for _, e := range events {
		if e.StartTime.Before(startTime) {
			startTime = e.StartTime
		}
		if e.EndTime.After(endTime) {
			endTime = e.EndTime
		}
	}

	duration := endTime.Sub(startTime)

	// Count components
	tikvCount := 0
	tiflashCount := 0
	causes := make(map[string]struct{})
	maxLevel := "minor"
	diskType := events[0].DiskType

	for _, e := range events {
		if e.Component == "tikv" {
			tikvCount++
		} else if e.Component == "tiflash" {
			tiflashCount++
		}

		// Collect root causes
		if e.EBS != nil {
			if e.EBS.ThroughputExceeded > 0 {
				causes["EBS throughput exceeded"] = struct{}{}
			}
			if e.EBS.IOPSExceeded > 0 {
				causes["EBS iops exceeded"] = struct{}{}
			}
		}

		// Track max anomaly level
		if levelWeight(e.AnomalyLevel) > levelWeight(maxLevel) {
			maxLevel = e.AnomalyLevel
		}
	}

	// Build impact scope
	impactScope := ""
	if tikvCount > 0 {
		impactScope = fmt.Sprintf("tikv %d", tikvCount)
	}
	if tiflashCount > 0 {
		if impactScope != "" {
			impactScope += ", "
		}
		impactScope += fmt.Sprintf("tiflash %d", tiflashCount)
	}

	// Build root cause
	rootCause := "Unknown cause"
	if len(causes) > 0 {
		causeList := make([]string, 0, len(causes))
		for c := range causes {
			causeList = append(causeList, c)
		}
		rootCause = strings.Join(causeList, ", ")
	}

	// Build jitter impact description
	jitterImpact := buildJitterImpact(events, params)

	return EventRecord{
		StartTime:              startTime,
		EndTime:                endTime,
		ClusterID:              clusterID,
		DiskType:               diskType,
		Duration:               duration.String(),
		JitterLevel:            maxLevel,
		ImpactScope:            impactScope,
		RootCause:              rootCause,
		JitterImpact:           jitterImpact,
		DiskMonitoringLink:     "", // Can be filled with Grafana URL if needed
		BusinessMonitoringLink: "", // Can be filled with Grafana URL if needed
	}
}

// buildJitterImpact builds the impact description
func buildJitterImpact(events []RawDiskAnomalyEvent, params DiskCheckParams) string {
	var hasTiDBImpact, hasTiFlashImpact bool
	var tidbRatio, tiflashRatio float64

	impactThreshold := params.ImpactRatioThreshold
	tiflashMinDuration := time.Duration(params.TiFlashMinDurationMin) * time.Minute

	for _, e := range events {
		if e.TiDB != nil && e.TiDB.Ratio > impactThreshold {
			hasTiDBImpact = true
			if e.TiDB.Ratio > tidbRatio {
				tidbRatio = e.TiDB.Ratio
			}
		}
		if e.TiFlash != nil && e.TiFlash.Ratio > impactThreshold && e.EndTime.Sub(e.StartTime) > tiflashMinDuration {
			hasTiFlashImpact = true
			if e.TiFlash.Ratio > tiflashRatio {
				tiflashRatio = e.TiFlash.Ratio
			}
		}
	}

	var parts []string
	if hasTiDBImpact {
		// Find baseline and peak for TiDB
		for _, e := range events {
			if e.TiDB != nil {
				baseline := int64(e.TiDB.Baseline * 1000)
				peak := int64(e.TiDB.Peak * 1000)
				parts = append(parts, fmt.Sprintf("TiDB P999 Duration increased (%d ms => %d ms)", baseline, peak))
				break
			}
		}
	}
	if hasTiFlashImpact {
		for _, e := range events {
			if e.TiFlash != nil {
				baseline := int64(e.TiFlash.Baseline * 1000)
				peak := int64(e.TiFlash.Peak * 1000)
				parts = append(parts, fmt.Sprintf("TiFlash P999 Run-mpp-Task Duration increased (%d ms => %d ms)", baseline, peak))

				raftWait := int64(e.TiFlash.RaftWaitIndexDuration * 1000)
				raftBatch := int64(e.TiFlash.RaftBatchReadIndexDuration * 1000)
				parts = append(parts, fmt.Sprintf("TiFlash P99 Raft Wait Index Duration: %d ms, Raft Batch Read Index Duration: %d ms", raftWait, raftBatch))
				break
			}
		}
	}

	return strings.Join(parts, "<br>")
}

// levelWeight returns the weight of anomaly level for comparison
func levelWeight(level string) int {
	switch level {
	case "critical":
		return 4
	case "major":
		return 3
	case "warning":
		return 2
	case "minor":
		return 1
	default:
		return 0
	}
}

// classifyEvent classifies an aggregated event
func classifyEvent(event EventRecord) EventClassification {
	classification := EventClassification{
		DiskType: event.DiskType,
	}

	// Check if TiKV or TiFlash
	if strings.Contains(event.ImpactScope, "tikv") {
		classification.IsTiKV = true
	}

	// Check if TiDB Issue (EBS exceeded) or Jitter Event (Unknown cause)
	if event.RootCause != "Unknown cause" {
		classification.IsTiDBIssue = true
	}

	// Check if impacted (has jitter impact description)
	if event.JitterImpact != "" {
		classification.IsImpacted = true
	}

	return classification
}

// buildStatistics builds the statistics table
func buildStatistics(events []EventRecord) []DiskStats {
	// Supported disk types
	diskTypes := []string{"io2", "gp3", "pd_ssd", "pd_balanced", "pd_extreme", "pd_standard"}

	// Initialize counters for each combination
	counters := make(map[string]map[string]map[string]int)
	for _, dt := range diskTypes {
		counters[dt] = map[string]map[string]int{
			"Impacted":     {"JitterEvents": 0, "TiKV": 0, "TiFlash": 0},
			"Non-Impacted": {"JitterEvents": 0, "TiKV": 0, "TiFlash": 0},
		}
	}

	for _, event := range events {
		classification := classifyEvent(event)

		impactType := "Non-Impacted"
		if classification.IsImpacted {
			impactType = "Impacted"
		}

		// Ensure disk type exists in counters
		diskType := classification.DiskType
		if _, ok := counters[diskType]; !ok {
			diskType = "gp3" // default fallback
		}

		if classification.IsTiDBIssue {
			if classification.IsTiKV {
				counters[diskType][impactType]["TiKV"]++
			} else {
				counters[diskType][impactType]["TiFlash"]++
			}
		} else {
			counters[diskType][impactType]["JitterEvents"]++
		}
	}

	// Build stats slice
	var stats []DiskStats
	for _, diskType := range diskTypes {
		for _, impactType := range []string{"Impacted", "Non-Impacted"} {
			stats = append(stats, DiskStats{
				DiskType:           diskType,
				ImpactType:         impactType,
				JitterEventsCount:  counters[diskType][impactType]["JitterEvents"],
				TiKVIssuesCount:    counters[diskType][impactType]["TiKV"],
				TiFlashIssuesCount: counters[diskType][impactType]["TiFlash"],
			})
		}
	}

	return stats
}

// countDisks counts disks by type and component
func countDisks(events []RawDiskAnomalyEvent) DiskCount {
	diskTypes := make(map[string]string)  // instance -> diskType
	components := make(map[string]string) // instance -> component

	for _, e := range events {
		diskTypes[e.InstanceName] = e.DiskType
		components[e.InstanceName] = e.Component
	}

	count := DiskCount{}
	for instance, diskType := range diskTypes {
		switch diskType {
		case "io2":
			count.IO2++
		case "gp3":
			count.GP3++
		case "pd_ssd":
			count.PDSSD++
		case "pd_balanced":
			count.PDBalanced++
		case "pd_extreme":
			count.PDExtreme++
		case "pd_standard":
			count.PDStandard++
		default:
			count.GP3++ // default fallback
		}

		if components[instance] == "tikv" {
			count.TiKV++
		} else if components[instance] == "tiflash" {
			count.TiFlash++
		}
	}

	return count
}

// parseTimeRange parses time range from request
func parseTimeRange(tr healthcheck.HealthCheckTimeRange) (time.Time, time.Time, error) {
	start, err := time.Parse(time.RFC3339, tr.Start)
	if err != nil {
		start, err = time.Parse("2006-01-02 15:04:05", tr.Start)
		if err != nil {
			return time.Time{}, time.Time{}, fmt.Errorf("invalid start time format: %w", err)
		}
	}

	end, err := time.Parse(time.RFC3339, tr.End)
	if err != nil {
		end, err = time.Parse("2006-01-02 15:04:05", tr.End)
		if err != nil {
			return time.Time{}, time.Time{}, fmt.Errorf("invalid end time format: %w", err)
		}
	}

	return start, end, nil
}

// detectAnomalyWindows detects anomaly windows from Prometheus metrics
func detectAnomalyWindows(promURL, metricName, promQL string, startTime, endTime time.Time, params DiskCheckParams) ([]DiskAnomalyWindow, error) {
	var windows []DiskAnomalyWindow
	startTs := startTime.Unix()
	endTs := endTime.Unix()

	// Query Prometheus
	data, err := queryMetrics(promQL, promURL, startTs, endTs)
	if err != nil {
		return nil, fmt.Errorf("query metrics failed: %w", err)
	}

	// Process each time series
	for _, series := range data.Data.Result {
		var currentWindow *DiskAnomalyWindow
		component := series.Metric["component"]
		instance := series.Metric["instance"]

		for _, value := range series.Values {
			timestamp := int64(value[0].(float64))
			valueStr := value[1].(string)
			metricValue, _ := strconv.ParseFloat(valueStr, 64)

			// Determine threshold based on disk type
			threshold, diskType := getThresholdForDiskType(series.Metric, params)

			// Check if exceeds threshold
			if metricValue > threshold {
				if currentWindow == nil {
					currentWindow = &DiskAnomalyWindow{
						StartTime:    time.Unix(timestamp, 0).UTC(),
						MetricName:   metricName,
						InstanceName: instance,
						Component:    component,
						MaxValue:     metricValue,
						DiskType:     diskType,
						Labels:       series.Metric,
					}
				} else {
					currentWindow.EndTime = time.Unix(timestamp, 0).UTC()
					if metricValue > currentWindow.MaxValue {
						currentWindow.MaxValue = metricValue
					}
				}
			} else if currentWindow != nil {
				// Window ended, save it
				windows = append(windows, *currentWindow)
				currentWindow = nil
			}
		}

		// Handle last window if exists
		if currentWindow != nil {
			windows = append(windows, *currentWindow)
		}
	}

	return windows, nil
}

// getThresholdForDiskType returns the appropriate threshold and disk type based on metric labels
func getThresholdForDiskType(labels map[string]string, params DiskCheckParams) (float64, string) {
	// Try to get disk type from labels first
	diskType := "gp3" // default
	if dt, ok := labels["disk_type"]; ok && dt != "" {
		diskType = dt
	}

	// Return threshold based on disk type
	switch diskType {
	case "io2":
		return params.IO2Threshold, diskType
	case "gp3":
		return params.GP3Threshold, diskType
	case "pd_ssd":
		return params.PDSSDThreshold, diskType
	case "pd_balanced":
		return params.PDBalancedThreshold, diskType
	case "pd_extreme":
		return params.PDExtremeThreshold, diskType
	case "pd_standard":
		return params.PDStandardThreshold, diskType
	default:
		return params.GP3Threshold, diskType
	}
}

// isIO2Disk checks if disk is IO2 type
func isIO2Disk(diskType string) bool {
	return diskType == "io2"
}

// mergeWindows merges read and write windows, grouping by instance and time
func mergeWindows(readWindows, writeWindows []DiskAnomalyWindow, params DiskCheckParams) []DiskAnomalyWindow {
	windowMap := make(map[string][]DiskAnomalyWindow)

	// Group windows by instance
	for _, w := range readWindows {
		key := w.InstanceName
		windowMap[key] = append(windowMap[key], w)
	}
	for _, w := range writeWindows {
		key := w.InstanceName
		windowMap[key] = append(windowMap[key], w)
	}

	// Merge overlapping windows for each instance
	var merged []DiskAnomalyWindow
	for _, windows := range windowMap {
		merged = append(merged, mergeInstanceWindows(windows, params)...)
	}

	return merged
}

// mergeInstanceWindows merges overlapping windows for a single instance
func mergeInstanceWindows(windows []DiskAnomalyWindow, params DiskCheckParams) []DiskAnomalyWindow {
	if len(windows) == 0 {
		return nil
	}

	// Sort by start time
	for i := 0; i < len(windows)-1; i++ {
		for j := i + 1; j < len(windows); j++ {
			if windows[i].StartTime.After(windows[j].StartTime) {
				windows[i], windows[j] = windows[j], windows[i]
			}
		}
	}

	var merged []DiskAnomalyWindow
	current := windows[0]
	mergeInterval := time.Duration(params.WindowMergeInterval) * time.Minute

	for i := 1; i < len(windows); i++ {
		// Check if windows overlap or are close (within merge interval)
		if windows[i].StartTime.Sub(current.EndTime) <= mergeInterval {
			// Merge windows
			if windows[i].EndTime.After(current.EndTime) {
				current.EndTime = windows[i].EndTime
			}
			if windows[i].MaxValue > current.MaxValue {
				current.MaxValue = windows[i].MaxValue
			}
		} else {
			merged = append(merged, current)
			current = windows[i]
		}
	}
	merged = append(merged, current)

	return merged
}

// buildRawAnomalyEvents builds complete raw anomaly events with additional metrics
func buildRawAnomalyEvents(dataPlaneURL, clusterURL, deployType, clusterID, physicalClusterID string, windows []DiskAnomalyWindow, params DiskCheckParams) []RawDiskAnomalyEvent {
	var events []RawDiskAnomalyEvent

	for _, window := range windows {
		event := RawDiskAnomalyEvent{
			StartTime:    window.StartTime,
			EndTime:      window.EndTime,
			InstanceName: window.InstanceName,
			Component:    window.Component,
			DiskType:     window.DiskType,
			AnomalyLevel: evaluateAnomalyLevel(window.MaxValue, window.DiskType, params),
		}

		// Set max latency based on metric type
		if window.MetricName == "disk_read_latency" {
			event.MaxReadLatency = window.MaxValue
		} else {
			event.MaxWriteLatency = window.MaxValue
		}

		// Query disk quota metrics (AWS EBS or GCP PD) from DataPlaneMetricsUrl
		ebsMetrics := queryDiskQuotaMetrics(dataPlaneURL, window, params)
		if ebsMetrics != nil {
			event.EBS = ebsMetrics
		}

		// Query TiDB/TiFlash impact based on component (from ClusterMetricsUrl)
		if window.Component == "tikv" {
			tidbImpact := queryTiDBImpact(clusterURL, deployType, clusterID, window.StartTime, window.EndTime, params)
			if tidbImpact != nil {
				event.TiDB = tidbImpact
			}
		} else if window.Component == "tiflash" {
			tiflashImpact := queryTiFlashImpact(clusterURL, deployType, physicalClusterID, window.StartTime, window.EndTime, params)
			if tiflashImpact != nil {
				event.TiFlash = tiflashImpact
			}
		}

		events = append(events, event)
	}

	return events
}

// evaluateAnomalyLevel evaluates anomaly level based on max value and disk type
func evaluateAnomalyLevel(maxValue float64, diskType string, params DiskCheckParams) string {
	threshold := params.GP3Threshold

	switch diskType {
	case "io2":
		threshold = params.IO2Threshold
	case "pd_ssd":
		threshold = params.PDSSDThreshold
	case "pd_balanced":
		threshold = params.PDBalancedThreshold
	case "pd_extreme":
		threshold = params.PDExtremeThreshold
	case "pd_standard":
		threshold = params.PDStandardThreshold
	}

	ratio := maxValue / threshold
	if ratio > params.CriticalLevelRatio {
		return "critical"
	}
	if ratio > params.MajorLevelRatio {
		return "major"
	}
	if ratio > params.WarningLevelRatio {
		return "warning"
	}
	return "minor"
}

// queryDiskQuotaMetrics queries cloud provider specific disk quota metrics
// For AWS: uses EBS CSI driver metrics
// For GCP: currently not supported (no equivalent quota exceeded metrics available)
func queryDiskQuotaMetrics(promURL string, window DiskAnomalyWindow, params DiskCheckParams) *EBSMetrics {
	// Skip if GCP disk type (no equivalent exceeded metrics in GCP)
	if isGCPDiskType(window.DiskType) {
		return nil
	}

	// Query AWS EBS metrics for AWS disk types
	return queryAWSEBSMetrics(promURL, window, params)
}

// isGCPDiskType checks if the disk type is a GCP PD type
func isGCPDiskType(diskType string) bool {
	switch diskType {
	case "pd_ssd", "pd_balanced", "pd_extreme", "pd_standard":
		return true
	default:
		return false
	}
}

// isAWSDiskType checks if the disk type is an AWS EBS type
func isAWSDiskType(diskType string) bool {
	return diskType == "io2" || diskType == "gp3"
}

// queryAWSEBSMetrics queries AWS EBS CSI specific metrics
func queryAWSEBSMetrics(promURL string, window DiskAnomalyWindow, params DiskCheckParams) *EBSMetrics {
	// Query node name from instance
	nodeQuery := fmt.Sprintf(`kube_pod_info{pod="%s"}`, window.InstanceName)
	nodeResult, err := queryMetrics(nodeQuery, promURL, window.StartTime.Unix()-1000, window.EndTime.Unix())
	if err != nil || len(nodeResult.Data.Result) == 0 {
		return nil
	}
	node := nodeResult.Data.Result[0].Metric["node"]
	if node == "" {
		return nil
	}

	// Query EBS metrics
	ebsBuffer := time.Duration(params.EBSBufferMinutes) * time.Minute
	baselineStart := window.StartTime.Add(-ebsBuffer)
	baselineEnd := window.EndTime.Add(ebsBuffer)

	// Throughput exceeded (CSI driver level)
	tpQuery := fmt.Sprintf(`rate(aws_ebs_csi_exceeded_tp_seconds_total{instance="%s"}[1m])`, node)
	tpResult, _ := queryMetrics(tpQuery, promURL, baselineStart.Unix(), baselineEnd.Unix())
	maxTP := extractMaxValue(tpResult)

	// Throughput exceeded (EC2 level)
	tpEc2Query := fmt.Sprintf(`rate(aws_ebs_csi_ec2_exceeded_tp_seconds_total{instance="%s"}[1m])`, node)
	tpEc2Result, _ := queryMetrics(tpEc2Query, promURL, baselineStart.Unix(), baselineEnd.Unix())
	maxTPEc2 := extractMaxValue(tpEc2Result)

	// IOPS exceeded (CSI driver level)
	iopsQuery := fmt.Sprintf(`rate(aws_ebs_csi_exceeded_iops_seconds_total{instance="%s"}[1m])`, node)
	iopsResult, _ := queryMetrics(iopsQuery, promURL, baselineStart.Unix(), baselineEnd.Unix())
	maxIOPS := extractMaxValue(iopsResult)

	// IOPS exceeded (EC2 level)
	iopsEc2Query := fmt.Sprintf(`rate(aws_ebs_csi_ec2_exceeded_iops_seconds_total{instance="%s"}[1m])`, node)
	iopsEc2Result, _ := queryMetrics(iopsEc2Query, promURL, baselineStart.Unix(), baselineEnd.Unix())
	maxIOPSEc2 := extractMaxValue(iopsEc2Result)

	// Throughput value
	throughputQuery := fmt.Sprintf(`sum by (instance) (rate(aws_ebs_csi_read_bytes_total{instance="%s"}[1m]) + rate(aws_ebs_csi_write_bytes_total{instance="%s"}[1m]))`, node, node)
	throughputResult, _ := queryMetrics(throughputQuery, promURL, baselineStart.Unix(), baselineEnd.Unix())
	maxThroughput := extractMaxValue(throughputResult)

	// Use max value between CSI and EC2 level metrics
	finalMaxTP := maxTP
	if maxTPEc2 > finalMaxTP {
		finalMaxTP = maxTPEc2
	}
	finalMaxIOPS := maxIOPS
	if maxIOPSEc2 > finalMaxIOPS {
		finalMaxIOPS = maxIOPSEc2
	}

	if finalMaxTP == 0 && finalMaxIOPS == 0 {
		return nil
	}

	return &EBSMetrics{
		ThroughputExceeded: finalMaxTP,
		IOPSExceeded:       finalMaxIOPS,
		Throughput:         maxThroughput,
	}
}

// queryTiDBImpact queries TiDB performance impact
func queryTiDBImpact(promURL, deployType, clusterID string, startTime, endTime time.Time, params DiskCheckParams) *TiDBImpact {
	// In mock mode, return preset ratio values directly
	if isMockEnabled() {
		return getMockTiDBImpact(startTime, endTime)
	}

	tidbDurationPromQL := getTiDBDurationPromQL(deployType, clusterID)

	baselineDuration := time.Duration(params.BaselineDurationHours) * time.Hour
	baselineStart := startTime.Add(-baselineDuration)
	baselineEnd := endTime.Add(baselineDuration)

	baselineResult, err := queryMetrics(tidbDurationPromQL, promURL, baselineStart.Unix(), baselineEnd.Unix())
	if err != nil {
		return nil
	}
	baseline := calculateAverage(baselineResult)

	peakBuffer := time.Duration(params.PeakBufferMinutes) * time.Minute
	peakResult, err := queryMetrics(tidbDurationPromQL, promURL, startTime.Add(-peakBuffer).Unix(), endTime.Add(peakBuffer).Unix())
	if err != nil {
		return nil
	}
	peak := extractMaxValue(peakResult)

	if baseline == 0 {
		return nil
	}

	return &TiDBImpact{
		Baseline: baseline,
		Peak:     peak,
		Ratio:    peak / baseline,
	}
}

// getMockTiDBImpact returns preset TiDBImpact based on time window
// Aligned with mock data timeline:
// 0-4:   ratio=3.0 (TiDB Issue + Impacted)
// 8-12:  ratio=1.2 (TiDB Issue + Non-Impacted)
// 35-39: ratio=2.5 (Jitter + Impacted)
// 43-47: ratio=1.1 (Jitter + Non-Impacted)
func getMockTiDBImpact(startTime, endTime time.Time) *TiDBImpact {
	baseTime := int64(1774540800) // 2026-03-27 00:00:00
	startIdx := int((startTime.Unix() - baseTime) / 30)
	endIdx := int((endTime.Unix() - baseTime) / 30)

	// Use the midpoint of the window to determine the phase
	midIdx := (startIdx + endIdx) / 2

	var ratio float64
	switch {
	case midIdx >= 0 && midIdx < 4:
		ratio = 3.0 // TiDB Issue + Impacted
	case midIdx >= 8 && midIdx < 12:
		ratio = 1.2 // TiDB Issue + Non-Impacted
	case midIdx >= 35 && midIdx < 39:
		ratio = 2.5 // Jitter + Impacted
	case midIdx >= 43 && midIdx < 47:
		ratio = 1.1 // Jitter + Non-Impacted
	default:
		ratio = 1.0 // normal
	}

	baseline := 0.05 // 50ms
	return &TiDBImpact{
		Baseline: baseline,
		Peak:     baseline * ratio,
		Ratio:    ratio,
	}
}

// getMockTiFlashImpact returns preset TiFlashImpact based on time window
// Uses the same ratio values as TiDB
func getMockTiFlashImpact(startTime, endTime time.Time) *TiFlashImpact {
	baseTime := int64(1774540800) // 2026-03-27 00:00:00
	startIdx := int((startTime.Unix() - baseTime) / 30)
	endIdx := int((endTime.Unix() - baseTime) / 30)

	// Use the midpoint of the window to determine the phase
	midIdx := (startIdx + endIdx) / 2

	var ratio float64
	switch {
	case midIdx >= 0 && midIdx < 4:
		ratio = 3.0 // TiDB Issue + Impacted
	case midIdx >= 8 && midIdx < 12:
		ratio = 1.2 // TiDB Issue + Non-Impacted
	case midIdx >= 35 && midIdx < 39:
		ratio = 2.5 // Jitter + Impacted
	case midIdx >= 43 && midIdx < 47:
		ratio = 1.1 // Jitter + Non-Impacted
	default:
		ratio = 1.0 // normal
	}

	baseline := 0.1 // 100ms
	return &TiFlashImpact{
		Baseline:                   baseline,
		Peak:                       baseline * ratio,
		Ratio:                      ratio,
		RaftWaitIndexDuration:      0.001,
		RaftBatchReadIndexDuration: 0.002,
	}
}

// queryTiFlashImpact queries TiFlash performance impact
func queryTiFlashImpact(promURL, deployType, physicalClusterID string, startTime, endTime time.Time, params DiskCheckParams) *TiFlashImpact {
	// In mock mode, return preset ratio values directly
	if isMockEnabled() {
		return getMockTiFlashImpact(startTime, endTime)
	}

	tiflashDurationPromQL := getTiFlashDurationPromQL(deployType, physicalClusterID)
	tiflashRaftWaitIndexPromQL := getTiFlashRaftWaitIndexPromQL(deployType, physicalClusterID)
	tiflashRaftBatchReadIndexPromQL := getTiFlashRaftBatchReadIndexPromQL(deployType, physicalClusterID)

	baselineDuration := time.Duration(params.BaselineDurationHours) * time.Hour
	baselineStart := startTime.Add(-baselineDuration)
	baselineEnd := endTime.Add(baselineDuration)

	baselineResult, err := queryMetrics(tiflashDurationPromQL, promURL, baselineStart.Unix(), baselineEnd.Unix())
	if err != nil {
		return nil
	}
	baseline := calculateAverage(baselineResult)

	peakBuffer := time.Duration(params.PeakBufferMinutes) * time.Minute
	peakResult, err := queryMetrics(tiflashDurationPromQL, promURL, startTime.Add(-peakBuffer).Unix(), endTime.Add(peakBuffer).Unix())
	if err != nil {
		return nil
	}
	peak := extractMaxValue(peakResult)

	raftBuffer := time.Duration(params.RaftBufferMinutes) * time.Minute
	raftStart := startTime.Add(-raftBuffer)
	raftEnd := endTime.Add(raftBuffer)

	raftWaitResult, _ := queryMetrics(tiflashRaftWaitIndexPromQL, promURL, raftStart.Unix(), raftEnd.Unix())
	raftWaitIndex := extractMaxValue(raftWaitResult)

	raftBatchResult, _ := queryMetrics(tiflashRaftBatchReadIndexPromQL, promURL, raftStart.Unix(), raftEnd.Unix())
	raftBatchReadIndex := extractMaxValue(raftBatchResult)

	return &TiFlashImpact{
		Baseline:                   baseline,
		Peak:                       peak,
		Ratio:                      safeDivide(peak, baseline),
		RaftWaitIndexDuration:      raftWaitIndex,
		RaftBatchReadIndexDuration: raftBatchReadIndex,
	}
}

// getDiskReadLatencyPromQL returns PromQL for disk read latency
func getDiskReadLatencyPromQL(deployType, clusterID string) string {
	if deployType == o11y.PremiumType || deployType == o11y.NgHostType {
		return fmt.Sprintf(`rate(node_disk_read_time_seconds_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash',tidb_cluster_id="%s"}[2m]) / rate(node_disk_reads_completed_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash',tidb_cluster_id="%s"}[2m])`, clusterID, clusterID)
	}
	return `rate(node_disk_read_time_seconds_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash'}[2m]) / rate(node_disk_reads_completed_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash'}[2m])`
}

// getDiskWriteLatencyPromQL returns PromQL for disk write latency
func getDiskWriteLatencyPromQL(deployType, clusterID string) string {
	if deployType == o11y.PremiumType || deployType == o11y.NgHostType {
		return fmt.Sprintf(`rate(node_disk_write_time_seconds_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash',tidb_cluster_id="%s"}[2m]) / rate(node_disk_writes_completed_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash',tidb_cluster_id="%s"}[2m])`, clusterID, clusterID)
	}
	return `rate(node_disk_write_time_seconds_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash'}[2m]) / rate(node_disk_writes_completed_total{device!~"dm-.*|nvme1.*|nvme0.*", component=~'tikv|tiflash'}[2m])`
}

// getTiDBDurationPromQL returns PromQL for TiDB duration
func getTiDBDurationPromQL(deployType, clusterID string) string {
	if deployType == o11y.PremiumType || deployType == o11y.NgHostType {
		return fmt.Sprintf(`histogram_quantile(0.999, sum(rate(tidb_server_handle_query_duration_seconds_bucket{tidb_cluster_id="%s"}[2m])) by (le))`, clusterID)
	}
	return `histogram_quantile(0.999, sum(rate(tidb_server_handle_query_duration_seconds_bucket{}[2m])) by (le))`
}

// getTiFlashDurationPromQL returns PromQL for TiFlash duration
func getTiFlashDurationPromQL(deployType, physicalClusterID string) string {
	if deployType == o11y.PremiumType || deployType == o11y.NgHostType {
		return fmt.Sprintf(`histogram_quantile(0.999, sum(rate(tiflash_coprocessor_request_duration_seconds_bucket{type="run_mpp_task",tidb_cluster_id="%s"}[2m])) by (le))`, physicalClusterID)
	}
	return `histogram_quantile(0.999, sum(rate(tiflash_coprocessor_request_duration_seconds_bucket{type="run_mpp_task"}[2m])) by (le))`
}

// getTiFlashRaftWaitIndexPromQL returns PromQL for TiFlash raft wait index duration
func getTiFlashRaftWaitIndexPromQL(deployType, physicalClusterID string) string {
	if deployType == o11y.PremiumType || deployType == o11y.NgHostType {
		return fmt.Sprintf(`histogram_quantile(0.99, sum(rate(tiflash_raft_wait_index_duration_seconds_bucket{tidb_cluster_id="%s"}[2m])) by (le))`, physicalClusterID)
	}
	return `histogram_quantile(0.99, sum(rate(tiflash_raft_wait_index_duration_seconds_bucket{}[2m])) by (le))`
}

// getTiFlashRaftBatchReadIndexPromQL returns PromQL for TiFlash raft batch read index duration
func getTiFlashRaftBatchReadIndexPromQL(deployType, physicalClusterID string) string {
	if deployType == o11y.PremiumType || deployType == o11y.NgHostType {
		return fmt.Sprintf(`histogram_quantile(0.99, sum(rate(tiflash_raft_batch_read_index_duration_seconds_bucket{tidb_cluster_id="%s"}[2m])) by (le))`, physicalClusterID)
	}
	return `histogram_quantile(0.99, sum(rate(tiflash_raft_batch_read_index_duration_seconds_bucket{}[2m])) by (le))`
}

// queryMetrics queries Prometheus metrics
func queryMetrics(query, promURL string, start, end int64) (*QueryResult, error) {
	// If mock is enabled, return constructed test data
	if isMockEnabled() {
		return mockQueryResult(query, promURL, start, end)
	}

	baseURL := promURL
	if !strings.HasSuffix(baseURL, "/") {
		baseURL += "/"
	}
	baseURL += "query_range"

	params := url.Values{}
	params.Set("query", query)
	params.Set("start", fmt.Sprintf("%d", start))
	params.Set("end", fmt.Sprintf("%d", end))
	params.Set("step", "30s")

	resp, err := http.Get(baseURL + "?" + params.Encode())
	if err != nil {
		return nil, fmt.Errorf("http request failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
	}

	var result QueryResult
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, fmt.Errorf("decode response failed: %w", err)
	}

	return &result, nil
}

// extractMaxValue extracts maximum value from query result
func extractMaxValue(result *QueryResult) float64 {
	if result == nil {
		return 0
	}
	var max float64
	for _, r := range result.Data.Result {
		for _, v := range r.Values {
			if value, ok := v[1].(string); ok {
				if f, err := strconv.ParseFloat(value, 64); err == nil && f > max {
					max = f
				}
			}
		}
	}
	return max
}

// calculateAverage calculates average value from query result
func calculateAverage(result *QueryResult) float64 {
	if result == nil || len(result.Data.Result) == 0 {
		return 0
	}
	var sum, count float64
	for _, r := range result.Data.Result {
		for _, v := range r.Values {
			if value, ok := v[1].(string); ok {
				if f, err := strconv.ParseFloat(value, 64); err == nil {
					sum += f
					count++
				}
			}
		}
	}
	if count == 0 {
		return 0
	}
	return sum / count
}

// safeDivide performs safe division
func safeDivide(a, b float64) float64 {
	if b == 0 {
		return 0
	}
	return a / b
}

// uploadToS3 uploads result to S3
func uploadToS3(key string, result *DiskHealthReport) error {
	region := os.Getenv("AWS_REGION")
	if region == "" {
		region = "us-west-2"
	}
	bucket := os.Getenv("HEALTHCHECK_S3_BUCKET")
	if bucket == "" {
		bucket = "delicated-healthcheck"
	}

	assumeRoleARN := os.Getenv("HEALTHCHECK_S3_ROLE_ARN")
	if assumeRoleARN == "" {
		assumeRoleARN = "arn:aws:iam::552185537300:role/clinic-job-role"
	}

	sess, err := session.NewSession(&aws.Config{
		Region: aws.String(region),
	})
	if err != nil {
		return fmt.Errorf("create s3 session failed: %w", err)
	}

	stsClient := sts.New(sess)
	creds := stscreds.NewCredentialsWithClient(stsClient, assumeRoleARN)

	s3Client := s3.New(sess, &aws.Config{Credentials: creds})
	uploader := s3manager.NewUploaderWithClient(s3Client)

	jsonData, err := json.MarshalIndent(result, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal result to json failed: %w", err)
	}

	_, err = uploader.Upload(&s3manager.UploadInput{
		Bucket:      aws.String(bucket),
		Key:         aws.String(key),
		Body:        bytes.NewReader(jsonData),
		ContentType: aws.String("application/json"),
	})
	if err != nil {
		return fmt.Errorf("upload to s3 failed: %w", err)
	}

	return nil
}