package s3

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/common"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/constant"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dao"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	log2 "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"
)

// S3Statistics S3 processing statistics
type S3Statistics struct {
	SlowQueryCount           int `json:"slow_query_count"`
	SlowQueryPatternCount    int `json:"slow_query_pattern_count"`
	SlowQueryAnalyzeCount    int `json:"slow_query_analyze_count"`
	SlowQueryDDLCount        int `json:"slow_query_ddl_count"`
	SlowQueryTiflashCount    int `json:"slow_query_tiflash_count"`
	SlowQueryForceIndexCount int `json:"slow_query_force_index_count"`
}

// LocalStatistics Local file processing statistics
type LocalStatistics struct {
	FileName                 string
	SlowQueryCount           int64
	SlowQueryPatternCount    int64
	SlowQueryAnalyzeCount    int64
	SlowQueryDDLCount        int64
	SlowQueryTiflashCount    int64
	SlowQueryForceIndexCount int64
	Error                    error
}

// S3LogParser S3 log parser
type S3LogParser struct {
	s3Client  *S3Client
	dbManager *dao.SQLiteManager
	config    *config.Config
	// Statistics variables
	slowQueryCount           int64
	slowQueryPatternCount    int64
	slowQueryAnalyzeCount    int64
	slowQueryDDLCount        int64
	SlowQueryTiflashCount    int64
	SlowQueryForceIndexCount int64
	// Concurrency protection
	mu sync.Mutex
}

// NewS3LogParser Create a new S3 log parser
func NewS3LogParser(s3Client *S3Client, dbManager *dao.SQLiteManager, config *config.Config) *S3LogParser {
	return &S3LogParser{
		s3Client:  s3Client,
		dbManager: dbManager,
		config:    config,
	}
}

// DownloadS3FilesFromConfig downloads files from S3 based on the configuration.
func DownloadS3FilesFromConfig(ctx context.Context) error {
	// orgs := &config.OrgsConfig
	cfg := &config.AppConfig

	orgClusters, err := common.ListOrgClusters(ctx)
	if err != nil {
		log2.Error(ctx, "ListEntities failed", zap.Error(err))
		return err
	}

	for orgId, clusters := range orgClusters {
		for _, cluster := range clusters {
			// respect skipCheck from config
			if cluster.SkipCheck {
				log2.Info(ctx, fmt.Sprintf("skip cluster %s (%s)", cluster.ClusterID, cluster.Alias))
				continue
			}
			// dbCluster, _ := cluster

			if cluster.Region == "" {
				return fmt.Errorf("cluster %s region is empty", cluster.ClusterID)
			}
			s3Client, err := GetS3ClientByGithubToken(cluster.Region)

			if err != nil {
				return fmt.Errorf("failed to get S3 client: %v", err)
			}
			// 1. download by day
			// s3Prefix := getClusterS3Prefix(cluster)
			// logFiles, err := s3Client.ListFiles(s3Prefix)

			// 2. download by time range
			logFiles, err := ListFilesByTimeRange(*s3Client, cluster.ClusterID)
			if err != nil {
				log2.PlainError("Failed to list S3 log files", zap.String("cluster", cluster.ClusterID), zap.Error(err))
				return err
			}
			if len(logFiles) == 0 {
				log2.Info(ctx, fmt.Sprintf("no log files found for cluster %s (%s)", cluster.ClusterID, cluster.Alias))
				continue
			}
			date := config.AppConfig.Day
			localDir := fmt.Sprintf("%s/%s/%s/%s", config.AppConfig.InputDir, date, orgId, cluster.ClusterID)
			parser := &S3LogParser{s3Client: s3Client, config: cfg}
			parser.downloadS3Files(ctx, logFiles, localDir)
		}
	}
	return nil
}

// func DownloadS3FilesByTimeRange(ctx context.Context, clusterId string, start, end time.Time) error {
// 	// clusterPtr := config.GetClusterByID(clusterId)
// 	// if clusterPtr == nil {
// 	// 	return fmt.Errorf("cluster %s not found", clusterId)
// 	// }
// 	// cluster := *clusterPtr

// 	// org := config.GetOrgByClusterID(cluster.ClusterID)
// 	dbCluster, _ := common.GetCluster(context.Background(), clusterId)

// 	if dbCluster == nil || dbCluster.Region == "" {
// 		return fmt.Errorf("cluster %s not found", clusterId)
// 	}
// 	s3Client, err := GetS3ClientByGithubToken(dbCluster.Region)

// 	if err != nil {
// 		return fmt.Errorf("failed to get S3 client: %v", err)
// 	}

// 	// 2. download by time range
// 	logFiles, err := ListFilesByTime(*s3Client, clusterId, start, end)
// 	if err != nil {
// 		log2.PlainError("Failed to list S3 log files", zap.String("cluster", clusterId), zap.Error(err))
// 		return err
// 	}
// 	if len(logFiles) == 0 {
// 		log2.PlainInfo(fmt.Sprintf("no log files found for cluster %s", clusterId))
// 		return nil
// 	}
// 	date := config.AppConfig.Day
// 	localDir := fmt.Sprintf("%s/%s/%s/%s", config.AppConfig.InputDir, date, dbCluster.OrgID, dbCluster.ClusterID)
// 	cfg := &config.AppConfig
// 	parser := &S3LogParser{s3Client: s3Client, config: cfg}
// 	parser.downloadS3Files(ctx, logFiles, localDir)

// 	return nil
// }

func getClusterS3Prefix(cluster dto.Cluster) string {
	// exple:"diagnosis/data/1379661944646416027/slowlog-merged/1h/2025070101"
	// exple:"diagnosis/data/1379661944646416027/slowlog-merged/1h/2025070102"
	date := config.AppConfig.Day
	s3Prefix := fmt.Sprintf("diagnosis/data/%s/slowlog-merged/1h/%s", cluster.ClusterID, date)
	return s3Prefix
}

func GetSlowQueryCheckTime(dayStr string) (start, end time.Time, err error) {

	// dayStr 格式为 "20250701"
	layout := "20060102"
	day, err := time.Parse(layout, dayStr)
	if err != nil {
		return time.Time{}, time.Time{}, err
	}
	//  4:00:00
	end = time.Date(day.Year(), day.Month(), day.Day(), 4, 0, 0, 0, time.Local)

	// yesterday 2:00:00
	start = end.Add(-23 * time.Hour)
	return start, end, nil
}

func ListFilesByTimeRange(s3Client S3Client, clusterId string) ([]string, error) {

	start, end, _ := GetSlowQueryCheckTime(config.AppConfig.Day)

	return ListFilesByTime(s3Client, clusterId, start, end)
}

func ListFilesByTime(s3Client S3Client, clusterId string, start, end time.Time) ([]string, error) {

	// start, end, _ := getStartEndTime(config.AppConfig.Day)

	var allFiles []string
	for t := start; !t.After(end); t = t.Add(time.Hour) {
		hourStr := t.Format("2006010215") // YYYYMMDDHH
		s3Prefix := fmt.Sprintf("diagnosis/data/%s/slowlog-merged/1h/%s", clusterId, hourStr)
		files, err := s3Client.ListFiles(s3Prefix)
		if err != nil {
			log2.PlainError("Failed to list S3 log files", zap.String("cluster", clusterId), zap.Error(err))
			return nil, err
		}
		allFiles = append(allFiles, files...)
	}
	return allFiles, nil
}

// downloadS3Files downloads S3 files to a local directory.
func (p *S3LogParser) downloadS3Files(ctx context.Context, logFiles []string, localDir string) error {
	// Create local download directory
	if err := os.MkdirAll(localDir, 0755); err != nil {
		return fmt.Errorf("failed to create local directory: %w", err)
	}

	var wg sync.WaitGroup
	// Download each file to local
	for i, file := range logFiles {
		wg.Add(1)
		go func(file string, i int) {
			defer wg.Done()
			log2.Info(ctx, fmt.Sprintf("Downloading file %d/%d: %s", i+1, len(logFiles), file))

			// Generate local file name
			fileName := filepath.Base(file)
			localPath := filepath.Join(localDir, fileName)
			// Check if file already exists
			if !utils.FileExists(localPath) {
				// log2.PlainInfo("File already exists, skipping download: " + localPath)
				// Download file
				if err := p.s3Client.DownloadFile(ctx, file, localPath); err != nil {
					log2.Error(ctx, "Failed to download file", zap.String("file", file), zap.Error(err))
					panic(err)
				}
				log2.Info(ctx, "File downloaded successfully: "+file+" -> "+localPath)
			} else {
				log2.Info(ctx, "File already exists, skipping download: "+localPath)
			}

		}(file, i)
	}
	wg.Wait()

	log2.Info(ctx, "S3 log files download completed, files saved in: "+localDir)
	return nil
}

// ListS3Files List all files with specified prefix in S3 bucket
func ListS3Files(cfg *config.Config) ([]string, error) {
	s3Client, err := GetS3Client(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to get S3 client: %v", err)
	}

	logFiles, err := s3Client.ListFiles(cfg.S3.Prefix)
	if err != nil {
		return nil, fmt.Errorf("failed to list S3 log files: %v", err)
	}

	log2.PlainInfo(fmt.Sprintf("Found %d S3 log files", len(logFiles)))
	return logFiles, nil
}

// processS3LogFile processes a single S3 log file (deprecated, recommend using processS3LogFileStream)
func (p *S3LogParser) processS3LogFile(key string) error {
	// Read file content from S3
	data, err := p.s3Client.ReadFile(key)
	if err != nil {
		return fmt.Errorf("failed to read file from S3: %w", err)
	}

	// Select parsing method based on file extension
	if strings.HasSuffix(strings.ToLower(key), ".json") {
		return p.parseJSONLogData(data, key)
	} else {
		// Assume it's a text format log
		return p.parseTextLogData(data, key)
	}
}

// parseJSONLogData parses JSON formatted log data
func (p *S3LogParser) parseJSONLogData(data []byte, filename string) error {
	var items []map[string]interface{}

	// Try to parse as a JSON array
	if err := json.Unmarshal(data, &items); err != nil {
		// If not an array, try to parse line by line (one JSON object per line)
		lines := strings.Split(string(data), "\n")
		for _, line := range lines {
			line = strings.TrimSpace(line)
			if line == "" {
				continue
			}

			var item map[string]interface{}
			if err := json.Unmarshal([]byte(line), &item); err != nil {
				log2.PlainError("Failed to parse JSON line", zap.String("filename", filename), zap.Error(err))
				continue
			}
			items = append(items, item)
		}
	}

	log2.PlainInfo(fmt.Sprintf("Parsed %d records from file %s", len(items), filename))

	// Process the parsed data
	return p.processLogItems(items, filename)
}

// parseTextLogData parses text-formatted log data
func (p *S3LogParser) parseTextLogData(data []byte, filename string) error {
	// Parsing can be done here based on the specific text log format
	// For now, simply assume each line is a JSON object
	lines := strings.Split(string(data), "\n")
	var items []map[string]interface{}

	for _, line := range lines {
		line = strings.TrimSpace(line)
		if line == "" || !strings.HasPrefix(line, "{") {
			continue
		}

		var item map[string]interface{}
		if err := json.Unmarshal([]byte(line), &item); err != nil {
			log2.PlainError("Failed to parse text log line", zap.String("filename", filename), zap.Error(err))
			continue
		}
		items = append(items, item)
	}

	log2.PlainInfo(fmt.Sprintf("Parsed %d records from text file %s", len(items), filename))

	return p.processLogItems(items, filename)
}

// parseTextLogLine parses a single line of text-formatted log data (maintaining original interface compatibility)
func (p *S3LogParser) parseTextLogLine(line string, filename string) error {
	logEntry, err := p.parseTextLogLineToStruct(line, filename)
	if err != nil {
		return err
	}
	if logEntry != nil {
		return p.dbManager.InsertLogEntry(*logEntry)
	}
	return nil
}

// [Production] parseTextLogLineToStruct parses a text-formatted log line and returns structured data
func (p *S3LogParser) parseTextLogLineToStruct(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)
	}
	timestamp := float64(timestampUs) / 1_000_000.0
	formattedTimestamp := fmt.Sprintf("%.6f", timestamp)

	// Parse JSON part
	var data map[string]interface{}
	if err := json.Unmarshal([]byte(jsonPart), &data); err != nil {
		log2.PlainError("JSON decode error", zap.String("filename", filename), zap.Error(err))
		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,
		"Timestamp_str":       utils.TimestampToTime(timestamp).UTC().Format("2006-01-02 15:04:05"),
		"Timestamp":           timestamp,
	}
	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 {
			log2.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
	}

	// Apply filter conditions and return struct
	return p.processAndReturnLogEntry(data, filename)
}

/* // processAndStoreLogEntry processes a single log entry (applies filter conditions and stores in DB)
func (p *S3LogParser) processAndStoreLogEntry(data map[string]interface{}, filename string) error {
	// Apply filter conditions
	if !p.shouldKeepLogEntry(data) {
		return nil // Skip if it does not meet the conditions
	}
	p.slowQueryPatternCount++
	// Convert to SlowQueryLog struct and store in DB
	if p.dbManager != nil {
		jsonBytes, err := json.Marshal(data)
		if err != nil {
			return fmt.Errorf("failed to marshal log entry to JSON: %w", err)
		}

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

		// Insert into database
		if err := p.dbManager.InsertLogEntry(logEntry); err != nil {
			return fmt.Errorf("failed to insert log entry into database: %w", err)
		}
	}

	return nil
}
*/
// processAndReturnLogEntry processes a single log entry and returns a struct (applies filter conditions)
func (p *S3LogParser) processAndReturnLogEntry(data map[string]interface{}, filename string) (*dao.SlowQueryLog, error) {
	// Apply filter conditions
	if !p.shouldKeepLogEntry(data) {
		return nil, nil // Skip if it does not meet the conditions
	}
	// p.slowQueryPatternCount++

	// Convert to SlowQueryLog struct
	jsonBytes, err := json.Marshal(data)
	if err != nil {
		log2.PlainError("Failed to marshal log entry to JSON", zap.String("filename", filename), zap.Error(err))
		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 {
		log2.PlainError("Failed to unmarshal JSON to SlowQueryLog", zap.String("filename", filename), zap.Error(err))
		return nil, fmt.Errorf("failed to unmarshal JSON to SlowQueryLog: %w", err)
	}

	return &logEntry, nil
}

// [Production]  shouldKeepLogEntry determines if a log entry meets the filter conditions (based on ShouldKeep logic in parse_json_file.go) [Production]
func (p *S3LogParser) shouldKeepLogEntry(item map[string]interface{}) bool {
	// Total query count
	p.slowQueryCount++

	// Skip condition set 1
	/*
		if Isinternal, ok := item["Is_internal"].(bool); ok {
			// Internal SQL
			if Isinternal {
				return false
			}
		}
	*/
	// Skip condition set 2
	if queryStr, ok := item["Query"].(string); ok {
		queryLower := strings.ToLower(strings.TrimSpace(queryStr))
		// Query contains "read_from_storage(tiflash"
		if strings.Contains(queryLower, "read_from_storage(tiflash") && !strings.Contains(queryLower, "/*+ read_from_storage(tiflash") {
			p.SlowQueryTiflashCount++
			return false
		}
		// Query contains "force index"
		if strings.Contains(queryLower, "force index") {
			p.SlowQueryForceIndexCount++
			return false
		}

		// SQL starting with analyze (case-insensitive)
		if strings.HasPrefix(strings.ToLower(strings.TrimSpace(queryStr)), "analyze") {
			p.slowQueryAnalyzeCount++
			return false
		}
		// Exclude DDL statements (CREATE, ALTER, DROP, TRUNCATE, etc.)
		ddlKeywords := []string{"create", "alter", "drop", "truncate", "rename"}
		for _, keyword := range ddlKeywords {
			if strings.HasPrefix(queryLower, keyword) {
				p.slowQueryDDLCount++
				return false
			}
		}
		// SQL not starting with select (case-insensitive)
		if !strings.HasPrefix(queryLower, "select") {
			return false
		}
	}

	//  HugeResourceUsage: Mem_max > 5G
	if memMaxVal, ok := item["Mem_max"].(float64); ok {
		if memMaxVal > 5*1024*1024*1024 {
			item["Pattern_name"] = constant.PatternHugeResourceUsage
			return true
		}
	}

	responseTotal, _ := item["Write_sql_response_total"].(float64)

	//  HugeResourceUsage: Query_time - Write_sql_response_total > 40s
	if queryTime, ok := item["Query_time"].(float64); ok {
		latency := queryTime - responseTotal

		if latency > 5 {
			item["Pattern_name"] = constant.PatternHugeResourceUsage
			if latency > 40 {
				item["Pattern_name_sub"] = constant.PatternHugeResourceUsageSubLatency
			}
			return true
		}
	}

	// Condition Failed: Succ is false
	if succ, ok := item["Succ"].(bool); ok {
		if !succ {
			item["Pattern_name"] = constant.PatternFailed
			return true
		}
	}

	// Condition TableFullScan: Recursively find TableFullScan_xxx + estRows > 10000 in Binary_plan_parsed
	if planParsed, ok := item["Binary_plan_parsed"].(map[string]interface{}); ok {
		for _, v := range planParsed {
			if p.hasTableFullScan(v) {
				item["Pattern_name"] = constant.PatternTableFullScan
				return true
			}
		}
	}
	// Condition IndexFullScan: Recursively find IndexFullScan_xxx + estRows > 10000 in Binary_plan_parsed
	if planParsed, ok := item["Binary_plan_parsed"].(map[string]interface{}); ok {
		for _, v := range planParsed {
			if p.hasIndexFullScan(v) {
				item["Pattern_name"] = constant.PatternIndexFullScan
				return true
			}
		}
	}

	// Condition DeepOffset: Query contains OFFSET and the number is greater than 100000
	if queryStr, ok := item["Query"].(string); ok {
		re := regexp.MustCompile(`OFFSET\s+(\d+)`)
		if match := re.FindStringSubmatch(queryStr); len(match) == 2 {
			if offset, err := strconv.Atoi(match[1]); err == nil && offset > 100000 {
				item["Pattern_name"] = constant.PatternDeepOffset
				p.slowQueryPatternCount++
				return true
			}
		}
	}

	// Condition ExcessiveCopTask: cop_task node exists under Binary_plan_parsed and num > 100
	if planParsed, ok := item["Binary_plan_parsed"].(map[string]interface{}); ok {
		for _, v := range planParsed {
			if p.hasCopTaskNumGt100(v) {
				item["Pattern_name"] = constant.PatternExcessiveCopTask
				p.slowQueryPatternCount++
				return true
			}
		}
	}

	// Condition Stats missing: Stats string contains "missing"
	if statsStr, ok := item["Stats"].(string); ok {
		if regexp.MustCompile(`missing`).MatchString(statsStr) {
			item["Pattern_name"] = constant.PatternStatsMissing
			p.slowQueryPatternCount++
			return true
		}
	}

	// Condition Tiflash Scan Region Imbalance: region_balance > 10 under Binary_plan_parsed
	if planParsed, ok := item["Binary_plan_parsed"].(map[string]interface{}); ok {
		for _, v := range planParsed {
			if p.hasTiflashScanRegionImbalance(v) {
				item["Pattern_name"] = constant.PatternTiflashScanRegionImbalance
				p.slowQueryPatternCount++
				return true
			}
		}
	}

	// Condition SendResponseSlow: Write_sql_response_total > 10s
	if responseTotal > 10 {
		item["Pattern_name"] = constant.PatternSendResponseSlow
		p.slowQueryPatternCount++
		return true
	}

	return true
}

// GetStatistics gets the statistics
func (p *S3LogParser) GetStatistics() (int64, int64, int64, int64, int64) {
	return p.slowQueryCount, p.slowQueryPatternCount, p.slowQueryAnalyzeCount, p.slowQueryDDLCount, p.SlowQueryForceIndexCount
}

// PrintStatistics prints the statistics
func (p *S3LogParser) PrintStatistics() {
	log2.PlainInfo("=== S3 Log Processing Statistics ===")
	log2.PlainInfo(fmt.Sprintf("TotalCount: %d", p.slowQueryCount))
	log2.PlainInfo(fmt.Sprintf("PatternCount: %d", p.slowQueryPatternCount))
	log2.PlainInfo(fmt.Sprintf("AnalyzeCount: %d", p.slowQueryAnalyzeCount))
	log2.PlainInfo(fmt.Sprintf("DDLCount: %d", p.slowQueryDDLCount))
	log2.PlainInfo(fmt.Sprintf("TiflashCount: %d", p.SlowQueryTiflashCount))
	log2.PlainInfo(fmt.Sprintf("ForceIndexCount: %d", p.SlowQueryForceIndexCount))

	// log2.PlainInfo(fmt.Sprintf("IgnoreCount: %d", p.slowQueryCount-p.slowQueryPatternCount))
}

// ProcessLocalS3Logs processes local S3 log files and returns statistics
func ProcessLocalS3Logs(ctx context.Context, orgId string, table ...string) (S3Statistics, error) {
	day := config.AppConfig.Day

	var stats S3Statistics
	tableName := day
	// Get local file list
	// dir := fmt.Sprintf("%s/%s/%s", config.AppConfig.InputDir, day, org.OrgID)
	dir := config.GetOrgS3FileDir(orgId)
	dbPath := config.GetDbPath(orgId)
	if len(table) > 0 {
		tableName = table[0]
		dbPath = config.GetTikvDbPath(orgId)

		dir = config.GetTikvOrgS3FileDir(orgId)

	}
	// dir is exist
	if _, err := os.Stat(dir); os.IsNotExist(err) {
		log2.Warn(ctx, dir+" not exist", zap.Error(err))
		return stats, nil
	}
	log2.PlainInfo("dbPath:" + dbPath)
	// Initialize database manager
	dbManager, err := dao.NewSQLiteManager(dbPath, tableName)
	if err != nil {
		log2.Error(ctx, "failed to create database manager ", zap.Error(err))
		return stats, fmt.Errorf("failed to create database manager: %v", err)
	}
	defer dbManager.Close()

	// Clear tables
	if err := dao.DropTable(dbPath, tableName); err != nil {
		log2.Error(ctx, "Warning: Failed to clear table", zap.Error(err))
	}
	// Create tables with gorm
	if err := dao.CreateSlowQueryTable(orgId, table...); err != nil {
		return stats, fmt.Errorf("failed to create table: %v", err)
	}

	// Create local file parser (no S3 client needed)
	parser := &S3LogParser{
		s3Client:  nil, // No S3 client needed for local processing
		dbManager: dbManager,
		config:    &config.AppConfig,
	}

	// if _, err := os.Stat(dir); os.IsNotExist(err) && config.IsDev() {
	// 	return stats, nil
	// }
	logFiles, err := getLocalLogFiles(dir)
	if err != nil {
		return stats, fmt.Errorf("failed to list local log files: %v", err)
	}

	log2.PlainInfo(fmt.Sprintf("Found %d local log files", len(logFiles)))
	if len(logFiles) == 0 {
		return stats, nil
	}

	// Process all files concurrently and write to SQLite database
	if err := parser.ProcessLocalLogFiles(logFiles); err != nil {
		return stats, fmt.Errorf("failed to process local log files concurrently: %v", err)
	}
	log2.Info(ctx, "Local log file processing completed")

	// Get statistics
	slowQueryCount, slowQueryPatternCount, slowQueryAnalyzeCount, slowQueryDDLCount, slowQueryForceIndexCount := parser.GetStatistics()
	stats = S3Statistics{
		SlowQueryCount:           int(slowQueryCount),
		SlowQueryPatternCount:    int(slowQueryPatternCount),
		SlowQueryAnalyzeCount:    int(slowQueryAnalyzeCount),
		SlowQueryDDLCount:        int(slowQueryDDLCount),
		SlowQueryForceIndexCount: int(slowQueryForceIndexCount),
	}

	// Print statistics
	log2.PlainInfo("=== Local S3 File Processing Statistics ===")
	parser.PrintStatistics()

	// Save statistics to JSON file
	statsJSON, err := json.MarshalIndent(stats, "", "  ")
	if err != nil {
		log2.Error(ctx, "Failed to marshal stats to JSON", zap.Error(err))
	} else {
		// Ensure output directory exists
		if err := os.MkdirAll("output", 0755); err != nil {
			log2.Error(ctx, "Failed to create output directory", zap.Error(err))
		} else {
			// Save to parse.json file
			if err := os.WriteFile("output/parse.json", statsJSON, 0644); err != nil {
				log2.Error(ctx, "Failed to write parse.json", zap.Error(err))
			} else {
				log2.Info(ctx, "Statistics saved to output/parse.json")
			}
		}
	}

	return stats, nil
}

// ProcessS3Logs main function for processing S3 log files
func ProcessS3Logs(config *config.Config, dbManager *dao.SQLiteManager) (S3Statistics, error) {
	stats := S3Statistics{}
	var err error

	// Get S3 log file list
	logFiles, err := ListS3Files(config)
	if err != nil {
		return stats, fmt.Errorf("failed to list S3 files: %w", err)
	}

	// Create S3 client
	var s3Client *S3Client

	s3Client, err = GetS3Client(config)

	if err != nil {
		log2.PlainError("Failed to create S3 client", zap.Error(err))
		return stats, fmt.Errorf("failed to create S3 client: %v", err)
	}

	// Create S3 log parser
	parser := NewS3LogParser(s3Client, dbManager, config)

	// Process each file individually in streaming mode
	// for i, file := range logFiles {
	// 	log2.PlainInfo(fmt.Sprintf("Processing file %d/%d: %s", i+1, len(logFiles), file))

	// 	if err := parser.ProcessS3LogFileStream(file); err != nil {
	// 		log2.PlainError("Failed to process file", zap.String("file", file), zap.Error(err))
	// 		// Continue processing other files without interrupting the entire process
	// 		continue
	// 	}
	// }

	// Process all files concurrently
	if len(logFiles) > 0 {
		if err := parser.ProcessS3LogFiles(logFiles); err != nil {
			return stats, fmt.Errorf("failed to process S3 log files concurrently: %v", err)
		}
	}

	log2.PlainInfo("S3 log file streaming processing completed")

	// Get statistics
	slowQueryCount, slowQueryPatternCount, slowQueryAnalyzeCount, slowQueryDDLCount, slowQueryForceIndexCount := parser.GetStatistics()
	stats = S3Statistics{
		SlowQueryCount:           int(slowQueryCount),
		SlowQueryPatternCount:    int(slowQueryPatternCount),
		SlowQueryAnalyzeCount:    int(slowQueryAnalyzeCount),
		SlowQueryDDLCount:        int(slowQueryDDLCount),
		SlowQueryForceIndexCount: int(slowQueryForceIndexCount),
	}

	// Print statistics
	log2.PlainInfo("\n=== S3 Processing Statistics ===")
	parser.PrintStatistics()

	return stats, nil
}

// hasTableFullScan recursively checks if TableFullScan exists and estRows > 10000
func (p *S3LogParser) hasTableFullScan(node interface{}) bool {
	nodeMap, ok := node.(map[string]interface{})
	if !ok {
		return false
	}

	name, _ := nodeMap["name"].(string)
	if matched, _ := regexp.MatchString(`TableFullScan_\d+`, name); matched {
		// skip tiflash
		storeType, _ := nodeMap["storeType"].(string)

		if est, ok := nodeMap["estRows"]; ok {
			var estVal float64
			switch v := est.(type) {
			case string:
				if f, err := strconv.ParseFloat(v, 64); err == nil {
					estVal = f
				}
			case float64:
				estVal = v
			}
			if estVal > 100000 && storeType != "tiflash" {
				return true
			}
		}
	}

	// Recursively traverse children
	if children, ok := nodeMap["children"].([]interface{}); ok {
		for _, child := range children {
			if p.hasTableFullScan(child) {
				return true
			}
		}
	}

	return false
}

// hasIndexFullScan recursively checks if IndexFullScan exists and estRows > 10000
func (p *S3LogParser) hasIndexFullScan(node interface{}) bool {
	nodeMap, ok := node.(map[string]interface{})
	if !ok {
		return false
	}

	name, _ := nodeMap["name"].(string)
	if matched, _ := regexp.MatchString(`IndexFullScan_\d+`, name); matched {
		// skip tiflash
		storeType, _ := nodeMap["storeType"].(string)

		if est, ok := nodeMap["estRows"]; ok {
			var estVal float64
			switch v := est.(type) {
			case string:
				if f, err := strconv.ParseFloat(v, 64); err == nil {
					estVal = f
				}
			case float64:
				estVal = v
			}
			if estVal > 100000 && storeType != "tiflash" {
				return true
			}
		}
	}

	// Recursively traverse children
	if children, ok := nodeMap["children"].([]interface{}); ok {
		for _, child := range children {
			if p.hasIndexFullScan(child) {
				return true
			}
		}
	}

	return false
}

// hasTiflashScanRegionImbalance checks the condition: In Binary_plan_parsed, the max/min ratio in region_balance of tiflash_scan in copExecInfo is > 10.
// Simple example: "copExecInfo": tiflash_scan:{region_balance:{instance_num: 4, max/min: 20/1=20.000000}...}
// Demo: "copExecInfo": "tiflash_task:{...}, tiflash_scan:{...region_balance:{instance_num: 4, max/min: 2/1=2.000000}, ...}",
func (p *S3LogParser) hasTiflashScanRegionImbalance(node interface{}) bool {
	nodeMap, ok := node.(map[string]interface{})
	if !ok {
		return false
	}

	// Check copExecInfo field
	if copExecInfo, ok := nodeMap["copExecInfo"].(string); ok {
		// Check if it contains tiflash_scan
		if strings.Contains(copExecInfo, "tiflash_scan") {
			// Use regex to extract the max/min ratio from region_balance in tiflash_scan
			// Match pattern: tiflash_scan:{...region_balance:{...max/min: number/number=number.number...}...}
			regex := regexp.MustCompile(`tiflash_scan:\{[^}]*region_balance:\{[^}]*max/min:\s*(\d+(?:\.\d+)?)/(\d+(?:\.\d+)?)=([\d\.]+)`)
			matches := regex.FindStringSubmatch(copExecInfo)
			if len(matches) >= 4 {
				// Extract the result of the max/min ratio
				if ratio, err := strconv.ParseFloat(matches[3], 64); err == nil {
					if ratio > 10 {
						return true
					}
				}
			}
		}
	}

	// Recursively traverse children
	if children, ok := nodeMap["children"].([]interface{}); ok {
		for _, child := range children {
			if p.hasTiflashScanRegionImbalance(child) {
				return true
			}
		}
	}

	return false
}

// hasCopTaskNumGt100 checks if a cop_task node exists and num > 100
func (p *S3LogParser) hasCopTaskNumGt100(node interface{}) bool {
	nodeMap, ok := node.(map[string]interface{})
	if !ok {
		return false
	}
	// Check rootGroupExecInfo or cop_task string
	if arr, ok := nodeMap["rootGroupExecInfo"].([]interface{}); ok {
		for _, s := range arr {
			if str, ok := s.(string); ok {
				re := regexp.MustCompile(`cop_task:\s*\{[^}]*num:\s*(\d+)`)
				if match := re.FindStringSubmatch(str); len(match) == 2 {
					if num, err := strconv.Atoi(match[1]); err == nil && num > 100 {
						return true
					}
				}
			}
		}
	}
	// Recurse children
	if children, ok := nodeMap["children"].([]interface{}); ok {
		for _, child := range children {
			if p.hasCopTaskNumGt100(child) {
				return true
			}
		}
	}
	return false
}

// Use the previously defined LocalStatistics struct

// ProcessS3LogFiles concurrently processes S3 log files
func (p *S3LogParser) ProcessS3LogFiles(logFiles []string) error {
	log2.PlainInfo(fmt.Sprintf("Starting to process %d S3 log files concurrently", len(logFiles)))

	// Limit the maximum number of concurrent goroutines to avoid excessive resource consumption
	numFiles := len(logFiles)
	if numFiles > 24 {
		log2.PlainInfo("Number of files exceeds 24, will process in batches")
		return fmt.Errorf("number of files exceeds 24")
	}

	log2.PlainInfo(fmt.Sprintf("Using one-to-one concurrency mode: %d files correspond to %d goroutines", numFiles, numFiles))

	// Create a result channel
	resultChan := make(chan LocalStatistics, len(logFiles))

	// Start a goroutine for each file directly
	for i, file := range logFiles {
		go func(workerID int, fileName string) {
			log2.PlainInfo(fmt.Sprintf("Goroutine-%d started processing file: %s", workerID, fileName))
			result := LocalStatistics{FileName: fileName}

			// Create an independent parser instance for statistics
			localParser := &S3LogParser{
				s3Client:  p.s3Client,
				dbManager: p.dbManager, // Shared database connection (SQLite WAL mode supports concurrent read and write)
				config:    p.config,    // Shared configuration
			}

			if err := localParser.processS3LogFile(fileName); err != nil {
				result.Error = err
				log2.PlainError(fmt.Sprintf("Goroutine-%d failed to process file", workerID), zap.String("fileName", fileName), zap.Error(err))
			} else {
				// Get local statistics results
				result.SlowQueryCount = localParser.slowQueryCount
				result.SlowQueryPatternCount = localParser.slowQueryPatternCount
				result.SlowQueryAnalyzeCount = localParser.slowQueryAnalyzeCount
				result.SlowQueryDDLCount = localParser.slowQueryDDLCount
				result.SlowQueryTiflashCount = localParser.SlowQueryTiflashCount
				result.SlowQueryForceIndexCount = localParser.SlowQueryForceIndexCount
				log2.PlainInfo(fmt.Sprintf("Goroutine-%d finished file %s: slow queries=%d", workerID, fileName, result.SlowQueryCount))
			}

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

	// Collect results and summarize statistics
	completedFiles := 0
	for i := 0; i < len(logFiles); i++ {
		result := <-resultChan
		completedFiles++

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

		// Summarize statistical data (no lock needed as it's serial summarization)
		p.slowQueryCount += result.SlowQueryCount
		p.slowQueryPatternCount += result.SlowQueryPatternCount
		p.slowQueryAnalyzeCount += result.SlowQueryAnalyzeCount
		p.slowQueryDDLCount += result.SlowQueryDDLCount
		p.SlowQueryTiflashCount += result.SlowQueryTiflashCount
		p.SlowQueryForceIndexCount += result.SlowQueryForceIndexCount

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

	log2.PlainInfo("All files processed!")
	// Print final statistics
	p.PrintStatistics()
	return nil
}

// processLogItems processes the parsed log items
func (p *S3LogParser) processLogItems(items []map[string]interface{}, filename string) error {
	processedCount := 0
	for _, item := range items {
		// Use existing filtering logic
		if parser.AbandonedShouldKeep(item) {
			// Add file source information
			item["Source_file"] = filename

			// Insert into database
			if err := p.dbManager.InsertSlowQueryRecord(item); err != nil {
				log2.PlainError("Failed to insert into database", zap.Error(err))
				continue
			}
			processedCount++
		}
	}

	log2.PlainInfo(fmt.Sprintf("File %s: processed %d valid records", filename, processedCount))
	return nil
}

// getLocalLogFiles gets the list of local log files
func getLocalLogFiles(inputDir string) ([]string, error) {
	var logFiles []string

	err := filepath.Walk(inputDir, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}

		// Skip directories
		if info.IsDir() {
			return nil
		}

		// Check file extension (supports .log, .txt, .json, etc.)
		ext := strings.ToLower(filepath.Ext(path))
		if ext == ".log" || ext == ".txt" || ext == ".json" || ext == "" {
			logFiles = append(logFiles, path)
		}

		return nil
	})

	if err != nil {
		return nil, fmt.Errorf("failed to walk directory %s: %w", inputDir, err)
	}

	return logFiles, nil
}

// ProcessLocalLogFiles concurrently processes local log files using a worker pool
func (p *S3LogParser) ProcessLocalLogFiles(logFiles []string) error {
	log2.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
	// log2.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 = p.config.Parser.Concurrency
	log2.PlainInfo("read maxWorkers ", zap.Int("maxWorkers", maxWorkers))

	log2.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 {
				log2.PlainInfo(fmt.Sprintf("Worker-%d started processing file: %s", workerID, fileName))
				result := LocalStatistics{FileName: fileName}

				// Create an independent parser instance for statistics
				localParser := &S3LogParser{
					s3Client:  nil,         // No S3 client needed for local processing
					dbManager: p.dbManager, // Shared database connection (SQLite WAL mode supports concurrent read and write)
					config:    p.config,    // Shared configuration
				}

				if err := localParser.processLocalLogFile(fileName); err != nil {
					result.Error = err
					log2.PlainError(fmt.Sprintf("Worker-%d failed to process file", workerID), zap.String("fileName", fileName), zap.Error(err))
				} else {
					// Get local statistics results
					result.SlowQueryCount = localParser.slowQueryCount
					result.SlowQueryPatternCount = localParser.slowQueryPatternCount
					result.SlowQueryAnalyzeCount = localParser.slowQueryAnalyzeCount
					result.SlowQueryDDLCount = localParser.slowQueryDDLCount
					result.SlowQueryTiflashCount = localParser.SlowQueryTiflashCount
					result.SlowQueryForceIndexCount = localParser.SlowQueryForceIndexCount
					log2.PlainInfo(fmt.Sprintf("Worker-%d finished file %s: slow queries=%d", workerID, fileName, result.SlowQueryCount))
				}

				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 {
			log2.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.slowQueryCount += result.SlowQueryCount
		p.slowQueryPatternCount += result.SlowQueryPatternCount
		p.slowQueryAnalyzeCount += result.SlowQueryAnalyzeCount
		p.slowQueryDDLCount += result.SlowQueryDDLCount
		p.SlowQueryTiflashCount += result.SlowQueryTiflashCount
		p.SlowQueryForceIndexCount += result.SlowQueryForceIndexCount
		p.mu.Unlock()

		// Display progress and performance information
		progress := float64(completedFiles) / float64(len(logFiles)) * 100
		log2.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))
	}

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

// processLocalLogFile processes a single local log file
func (p *S3LogParser) processLocalLogFile(filePath string) error {
	log2.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
	batchSize := p.config.Parser.BatchSize
	if batchSize <= 0 {
		batchSize = 5000
	}
	batch := make([]dao.SlowQueryLog, 0, batchSize)

	flushBatch := func() error {
		if len(batch) == 0 {
			return nil
		}
		if err := p.dbManager.BatchInsertLogEntries(batch); err != nil {
			return fmt.Errorf("batch insert failed: %w", err)
		}
		batch = batch[:0]
		return nil
	}

	reader := bufio.NewReader(file)
	for {
		line, err := reader.ReadString('\n')
		if err != nil {
			if err == io.EOF {
				break
			}
			return fmt.Errorf("reading file %s: %w", filePath, err)
		}
		if strings.TrimSpace(line) == "" {
			continue
		}

		processedCount++

		logEntry, err := p.parseTextLogLineToStruct(line, filePath)
		if err != nil {
			errorCount++
			log2.PlainError(fmt.Sprintf("Failed to parse local file line (line %d)，filePath: %s", processedCount, filePath), zap.Any("line", line), zap.Error(err))
			continue
		}

		if logEntry != nil {
			pathParts := strings.Split(filePath, "/")
			if len(pathParts) >= 4 {
				logEntry.ClusterId = pathParts[len(pathParts)-2]
			}
			batch = append(batch, *logEntry)
			if len(batch) >= batchSize {
				if err := flushBatch(); err != nil {
					return fmt.Errorf("batch insert failed: %w", err)
				}
			}
		}

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

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

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

	return nil
}

// ProcessS3LogFilesByPattern processes S3 log files by pattern
func (p *S3LogParser) ProcessS3LogFilesByPattern(prefix string, pattern string, day string) error {
	logFiles, err := p.s3Client.ListFiles(prefix)
	if err != nil {
		return fmt.Errorf("failed to list log files: %w", err)
	}

	var matchedFiles []string
	for _, file := range logFiles {
		if strings.Contains(file, pattern) {
			matchedFiles = append(matchedFiles, file)
		}
	}

	log2.PlainInfo(fmt.Sprintf("Found %d matching files for pattern '%s'", len(matchedFiles), pattern))

	for _, file := range matchedFiles {
		log2.PlainInfo("Processing matching file: " + file)
		if err := p.processS3LogFile(file); err != nil {
			log2.PlainError("Failed to process file", zap.String("file", file), zap.Error(err))
			continue
		}
	}

	return nil
}