package s3

import (
	"bufio"
	"encoding/json"
	"fmt"
	"os"
	"regexp"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dao"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/parser"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/utils"
	"go.uber.org/zap"
)

type DBParser struct {
	ClusterName string
	ClusterId   string
	Day         string
	TotalDB     int64
	Top5        []LogBean
	// 统计字段

	// Concurrency protection
	mu sync.Mutex
}

type LogBean struct {
	DB         string
	TotalCount int64
	Precent    string
}

// DB统计结果结构体
type DBStat struct {
	SchemaName     string `json:"SchemaName"`
	TotalExecCount int64  `json:"TotalExecCount"`
	Precent        string `json:"Precent"`
}

// 集群DB统计结果
type ClusterDBResult struct {
	ClusterName    string   `json:"ClusterName"`
	ClusterId      string   `json:"ClusterId"`
	Day            string   `json:"Day"`
	TotalExecCount int64    `json:"TotalExecCount"`
	TotalDB        int64    `json:"TotalDB"`
	Top5           []DBStat `json:"Top5"`
	DB             []DBStat `json:"DB"`
}

type ProcessDBResult struct {
	ClusterStats map[string]ClusterDBResult `json:"ClusterStats"`
	Errors       map[string]string          `json:"Errors"`
}

// ProcessDB processes local S3 log files and returns DB statistics
func ProcessDB(clusterIdArr []string, day string) (ProcessDBResult, error) {
	result := ProcessDBResult{
		ClusterStats: make(map[string]ClusterDBResult),
		Errors:       make(map[string]string),
	}
	var mu sync.Mutex
	var wg sync.WaitGroup

	for _, clusterId := range clusterIdArr {
		wg.Add(1)
		go func(cid string) {
			defer wg.Done()
			cluster := config.GetClusterByID(cid)
			org := config.GetOrgByClusterID(cid)

			dir := fmt.Sprintf("%s/%s/%s/%s", config.AppConfig.InputDir, day, org.OrgID, cluster.ClusterID)
			// dir:=fmt.Sprintf("/Users/wlt/Downloads/slow-query/s3-file/%s/%s/%s", day,org.OrgID,cid)
			if _, err := os.Stat(dir); os.IsNotExist(err) {
				mu.Lock()
				result.Errors[cid] = "目录不存在"
				mu.Unlock()
				return
			}

			logFiles, err := getLocalLogFiles(dir)
			if err != nil {
				mu.Lock()
				result.Errors[cid] = err.Error()
				mu.Unlock()
				return
			}

			var dbList []string
			for _, file := range logFiles {
				items, err := GetDBItemsFromLogs(file)
				if err != nil {
					log.PlainError("Failed to parse log file", zap.String("file", file), zap.Error(err))
					continue
				}
				dbList = append(dbList, items...)
			}

			// 统计DB占比
			allStats := StatDBPercent(dbList)
			top5Stats := getTop5WithOther(allStats)
			// 构建结果
			clusterResult := ClusterDBResult{
				ClusterName:    cluster.Alias,
				ClusterId:      cid,
				Day:            day,
				TotalExecCount: getTotalCount(allStats),
				TotalDB:        int64(len(allStats)),
				Top5:           top5Stats,
				// DB: allStats,
			}

			mu.Lock()
			result.ClusterStats[cluster.Alias] = clusterResult
			mu.Unlock()
		}(clusterId)

	}

	wg.Wait()
	fileName := fmt.Sprintf("output/statdb/%s_top5.json", day)
	utils.WriteFile(fileName, result)
	return result, nil
}

// ProcessLocalLogFiles concurrently processes local log files using a worker pool
func (p *DBParser) ProcessFiles(logFiles []string) error {
	log.PlainInfo(fmt.Sprintf("Starting to concurrently process %d local log files", len(logFiles)))

	// Determine the number of workers based on the number of CPU cores
	maxWorkers := runtime.NumCPU() * 2
	// log.PlainInfo("read maxWorkers ", zap.Int("maxWorkers", maxWorkers))
	fileNum := len(logFiles)
	if fileNum <= 24 {
		maxWorkers = fileNum // Limit the maximum number of workers
	} else if maxWorkers > 14 {
		maxWorkers = 24
	} else if maxWorkers <= 14 {
		maxWorkers = 20 // Ensure a minimum number of workers
	}

	maxWorkers = 20
	log.PlainInfo("read maxWorkers ", zap.Int("maxWorkers", maxWorkers))

	log.PlainInfo(fmt.Sprintf("Using worker pool mode: %d files will be processed by %d workers", len(logFiles), maxWorkers))

	// Create file task channel and result channel
	fileChan := make(chan string, len(logFiles))
	resultChan := make(chan LocalStatistics, len(logFiles))

	// Start a fixed number of worker goroutines
	var wg sync.WaitGroup
	for i := 0; i < maxWorkers; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			for fileName := range fileChan {
				log.PlainInfo(fmt.Sprintf("Worker-%d started processing file: %s", workerID, fileName))
				result := LocalStatistics{FileName: fileName}

				// Create an independent parser instance for statistics
				localParser := &DBParser{
					ClusterName: p.ClusterName,
					ClusterId:   p.ClusterId,
					Day:         p.Day,
				}

				if err := localParser.processLocalLogFile(fileName); err != nil {
					result.Error = err
					log.PlainError(fmt.Sprintf("Worker-%d failed to process file", workerID), zap.String("fileName", fileName), zap.Error(err))
				}

				resultChan <- result
			}
		}(i + 1)
	}

	// Send all file tasks to the channel
	go func() {
		for _, file := range logFiles {
			fileChan <- file
		}
		close(fileChan) // Close the task channel to notify workers that there are no more tasks
	}()

	// Close the result channel after all workers are done
	go func() {
		wg.Wait()
		close(resultChan)
	}()

	// Collect results and summarize statistics
	completedFiles := 0
	for result := range resultChan {
		completedFiles++

		if result.Error != nil {
			log.PlainError(fmt.Sprintf("[%d/%d] Error processing file", completedFiles, len(logFiles)), zap.String("fileName", result.FileName), zap.Error(result.Error))
			continue
		}

		// Summarize statistical data (requires locking to protect concurrent access)
		// p.mu.Lock()
		// p.mu.Unlock()

		// Display progress and performance information
		progress := float64(completedFiles) / float64(len(logFiles)) * 100
		log.PlainInfo(fmt.Sprintf("[%d/%d] (%.1f%%) Completed file %s: slow queries=%d, pattern matches=%d",
			completedFiles, len(logFiles), progress, result.FileName,
			result.SlowQueryCount, result.SlowQueryPatternCount))
	}

	log.PlainInfo("All files processing completed!")
	// Print final statistics
	p.PrintStatistics()
	return nil
}

// processLocalLogFile processes a single local log file
func (p *DBParser) processLocalLogFile(filePath string) error {
	log.PlainInfo("Starting to process local file: " + filePath)

	file, err := os.Open(filePath)
	if err != nil {
		return fmt.Errorf("failed to open file %s: %w", filePath, err)
	}
	defer file.Close()

	processedCount := 0
	errorCount := 0
	// Get batch size from config, default is 1000
	batchSize := config.AppConfig.Parser.BatchSize
	if batchSize <= 0 {
		batchSize = 5000
	}
	// if config.IsDev() {
	// 	batchSize = 500000
	// }

	batch := make([]dao.SlowQueryLog, 0, batchSize)

	// Batch insert function
	flushBatch := func() error {
		if len(batch) == 0 {
			return nil
		}

		batch = batch[:0] // Clear the batch but keep the capacity
		return nil
	}

	scanner := bufio.NewScanner(file)
	// Set a larger buffer to handle long lines
	const maxScanTokenSize = 20 * 1024 * 1024 // 10MB
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, maxScanTokenSize)
	for scanner.Scan() {
		line := scanner.Text()
		if strings.TrimSpace(line) == "" {
			continue // Skip empty lines
		}

		processedCount++

		// Parse the log line and get structured data
		logEntry, err := p.parseLogLineToStruct(line, filePath)
		if err != nil {
			errorCount++
			log.PlainError(fmt.Sprintf("Failed to parse local file line (line %d)，filePath: %s", processedCount, filePath), zap.Error(err))
			continue // Continue to process other lines
		}

		// If parsing is successful and the filter conditions are met, add to the batch
		if logEntry != nil {

			// If parsing is successful and logEntry is not nil, extract ClusterId from filePath
			// filePath: ../s3-file/20250701/1372813089209061633/10218000100066887827/2025070101
			pathParts := strings.Split(filePath, "/")
			if len(pathParts) >= 4 {
				// ClusterId is the second to last part of the path
				logEntry.ClusterId = pathParts[len(pathParts)-2]
			}

			batch = append(batch, *logEntry)

			// When the batch reaches the specified size, execute a batch insert
			if len(batch) >= batchSize {
				if err := flushBatch(); err != nil {
					return fmt.Errorf("batch insert failed: %w", err)
				}
			}
		}

		// Output progress every 1000 lines
		if processedCount%1000 == 0 {
			log.PlainInfo(fmt.Sprintf("Processed %d lines, errors %d, batch size %d", processedCount, errorCount, len(batch)))
		}
	}

	if err := scanner.Err(); err != nil {
		return fmt.Errorf("reading file %s: %w", filePath, err)
	}

	// Process the remaining batch
	if err := flushBatch(); err != nil {
		panic(fmt.Errorf("final batch insert failed: %w", err))
	}

	log.PlainInfo(fmt.Sprintf("Local file %s processed: total %d lines, errors %d", filePath, processedCount, errorCount))

	return nil
}

func (p *DBParser) parseLogLineToStruct(line string, filename string) (*dao.SlowQueryLog, error) {
	// Skip comment lines
	if strings.HasPrefix(line, "#") {
		return nil, nil
	}

	// Extract timestamp and JSON part
	timestampMatch := regexp.MustCompile(`^(\d+)(.*)`).FindStringSubmatch(line)
	if len(timestampMatch) != 3 {
		return nil, fmt.Errorf("no timestamp or JSON found in line")
	}

	timestampStr := timestampMatch[1]
	jsonPart := timestampMatch[2]

	// Convert microsecond timestamp to second timestamp with decimal point
	timestampUs, err := strconv.ParseInt(timestampStr, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("invalid timestamp: %v", err)
	}
	formattedTimestamp := fmt.Sprintf("%.6f", float64(timestampUs)/1_000_000.0)

	// Parse JSON part
	var data map[string]interface{}
	if err := json.Unmarshal([]byte(jsonPart), &data); err != nil {
		return nil, fmt.Errorf("JSON decode error: %v", err)
	}

	// Insert the formatted timestamp at the beginning of the data
	newData := map[string]interface{}{
		"Formatted_Timestamp": formattedTimestamp,
		// "Cluster_Id": p.clusterId,
	}
	for key, value := range data {
		newData[key] = value
	}
	data = newData

	// Parse Binary_plan
	if binaryPlan, ok := data["Binary_plan"].(string); ok && binaryPlan != "" {
		if parsedPlan, err := parser.ParseBinaryPlan(binaryPlan); err != nil {
			log.PlainError("Error parsing Binary_plan", zap.Error(err))
			data["Binary_plan_parsed"] = nil
		} else {
			data["Binary_plan_parsed"] = parsedPlan
		}
	} else {
		data["Binary_plan_parsed"] = nil
	}

	// Convert to SlowQueryLog struct
	jsonBytes, err := json.Marshal(data)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal log entry to JSON: %w", err)
	}

	var logEntry dao.SlowQueryLog
	if err := json.Unmarshal(jsonBytes, &logEntry); err != nil {
		return nil, fmt.Errorf("failed to unmarshal JSON to SlowQueryLog: %w", err)
	}

	return &logEntry, nil

}

func StatDBPercent(dbItems []string) []DBStat {
	dbCount := make(map[string]int64)
	total := int64(0)
	for _, db := range dbItems {
		dbCount[db]++
		total++
	}
	var stats []DBStat
	for db, cnt := range dbCount {
		percent := float64(cnt) / float64(total) * 100
		stats = append(stats, DBStat{
			SchemaName:     db,
			TotalExecCount: cnt,
			Precent:        fmt.Sprintf("%.2f%%", percent),
		})
	}
	sort.Slice(stats, func(i, j int) bool {
		return stats[i].TotalExecCount > stats[j].TotalExecCount
	})
	return stats
}

// getTop5WithOther 获取前5个和其他的统计
func getTop5WithOther(stats []DBStat) []DBStat {
	if len(stats) <= 5 {
		return stats
	}

	top5 := make([]DBStat, 5)
	copy(top5, stats[:5])

	// 计算other的总数
	var otherCount int64
	for i := 5; i < len(stats); i++ {
		otherCount += stats[i].TotalExecCount
	}

	// 计算总数
	var totalCount int64
	for _, stat := range stats {
		totalCount += stat.TotalExecCount
	}

	// 添加other项
	if otherCount > 0 {
		otherPercent := float64(otherCount) / float64(totalCount) * 100
		other := DBStat{
			SchemaName:     "other",
			TotalExecCount: otherCount,
			Precent:        fmt.Sprintf("%.2f%%", otherPercent),
		}
		top5 = append(top5, other)
	}

	return top5
}

// getTotalCount 获取总执行次数
func getTotalCount(stats []DBStat) int64 {
	var total int64
	for _, stat := range stats {
		total += stat.TotalExecCount
	}
	return total
}

// GetDBItemsFromLogs 从日志文件中提取DB名称列表
func GetDBItemsFromLogs(filePath string) ([]string, error) {
	file, err := os.Open(filePath)
	if err != nil {
		return nil, err
	}
	defer file.Close()

	var dbList []string
	scanner := bufio.NewScanner(file)
	// 增加缓冲区大小以处理长行
	const maxScanTokenSize = 20 * 1024 * 1024 // 20MB
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, maxScanTokenSize)

	for scanner.Scan() {
		line := scanner.Text()
		line = strings.TrimSpace(line)

		// 跳过空行和注释行
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}

		// 提取时间戳后的JSON部分
		timestampMatch := regexp.MustCompile(`^(\d+)(.*)`).FindStringSubmatch(line)
		if len(timestampMatch) == 3 {
			jsonPart := timestampMatch[2]

			// 尝试解析JSON格式的日志行
			var logEntry map[string]interface{}
			if err := json.Unmarshal([]byte(jsonPart), &logEntry); err == nil {
				// 从JSON中提取DB字段
				if dbValue, exists := logEntry["DB"]; exists {
					if dbName, ok := dbValue.(string); ok && dbName != "" {
						dbList = append(dbList, dbName)
					}
				}
			}
		}

	}

	return dbList, scanner.Err()
}

// PrintStatistics 打印统计信息
func (p *DBParser) PrintStatistics() {
	log.PlainInfo("DB Statistics",
		zap.String("cluster", p.ClusterName),
		zap.String("clusterId", p.ClusterId),
		zap.Int64("totalDB", p.TotalDB),
	)
}