package main

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"context"
	"crypto/md5"
	"encoding/csv"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"os"
	"os/signal"
	"path/filepath"
	"regexp"
	"runtime"
	"runtime/debug"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"

	"net/http"

	oss "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	openapicred "github.com/aliyun/credentials-go/credentials"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	_ "github.com/mattn/go-sqlite3"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"golang.org/x/sync/semaphore"
)

var (
	env                 string
	vendor              string
	region              string
	srcBucket           string
	dstBucket           string
	logLevel            string
	s3concurrency       int64
	confPath            string
	srcPrefix           string
	shareDataPrefix     string
	initialMarkerPrefix int64
	maxFilesPerBatch    int
	maxOutputFileSize   int64
	parallelWorkers     int
	// enableRuV2 controls whether statement CSV keeps ru_v2 columns
	// (sum_ruv2/max_ruv2/AVG_REQUEST_UNIT_V2/MAX_REQUEST_UNIT_V2). Default false omits them.
	enableRuV2 bool
)

// defaultMaxFilesPerBatch is the max files per batch; flush and update checkpoint to avoid OOM on first full run.
const defaultMaxFilesPerBatch = 2000
const initialMarkerTrimDigits int64 = 10000

type ObjectConfig struct {
	Provider   string
	Region     string
	BucketName string
}

var (
	s3sem   *semaphore.Weighted
	timeNow = time.Now

	recordsRead = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "log_processor_records_read_total",
		Help: "Total number of log records read",
	})

	recordsWritten = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "log_processor_records_written_total",
		Help: "Total number of log records successfully written",
	})

	recordsMismatch = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "log_processor_records_mismatch_total",
		Help: "Total number of records where read count != written count",
	})

	processingErrors = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "log_processor_errors_total",
		Help: "Total number of processing errors",
	})
)

func init() {
	prometheus.MustRegister(recordsRead)
	prometheus.MustRegister(recordsWritten)
	prometheus.MustRegister(recordsMismatch)
	prometheus.MustRegister(processingErrors)

	if !testing.Testing() {
		flag.StringVar(&env, "env", "prod", "env")
		flag.StringVar(&vendor, "vendor", "alicloud", "vendor")
		flag.StringVar(&region, "region", "ap-southeast-1", "region")
		flag.StringVar(&srcBucket, "srcBucket", "sbucket", "source bucket")
		flag.StringVar(&dstBucket, "dstBucket", "dbucket", "dest bucket")
		flag.Int64Var(&s3concurrency, "s3-concurrency", 100, "s3 download concurrency")
		flag.StringVar(&confPath, "configPath", "/tmp/config.json", "confi json")
		flag.StringVar(&logLevel, "logLevel", "info", "log level")
		flag.StringVar(&srcPrefix, "srcPrefix", "0/prod-devtier/serverless", "source prefix")
		flag.StringVar(&shareDataPrefix, "shareDataPrefix", "share_data", "share data prefix for meta files")
		flag.Int64Var(&initialMarkerPrefix, "initial-marker-prefix", 0, "initial listing prefix when no marker exists (unix_ts/10000, e.g. 170902); 0 means auto-round current time")
		flag.IntVar(&maxFilesPerBatch, "max-files-per-batch", defaultMaxFilesPerBatch, "max files per batch; flush and update checkpoint to avoid OOM on first full run")
		flag.Int64Var(&maxOutputFileSize, "max-output-file-size", defaultMaxOutputFileSize, "max compressed output file size in bytes before rolling to a new file")
		flag.IntVar(&parallelWorkers, "parallel-workers", defaultParallelWorkers, "parallel download workers per batch")
		flag.BoolVar(&enableRuV2, "enable-ru-v2", false, "include statement ru_v2 CSV columns (sum_ruv2/max_ruv2/AVG_REQUEST_UNIT_V2/MAX_REQUEST_UNIT_V2); default false omits them")

		flag.Parse()
		s3sem = semaphore.NewWeighted(s3concurrency)

		http.Handle("/metrics", promhttp.Handler())
		srv := &http.Server{
			Addr:         ":8000",
			ReadTimeout:  5 * time.Second,
			WriteTimeout: 10 * time.Second,
			IdleTimeout:  15 * time.Second,
		}
		go func() {
			if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
				log.Error("Failed to start Prometheus metrics server", zap.Error(err))
			}
		}()
	}
}

func NewObjectStorage(ctx context.Context, cfg *ObjectConfig) (storage.ListStorage, error) {
	switch cfg.Provider {
	case "aws":
		cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(cfg.Region))
		if err != nil {
			return nil, err
		}
		return storage.NewAwsStorage(s3.NewFromConfig(cfg)), nil
	case "alicloud":
		cred, err := openapicred.NewCredential(nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create default AliCloud credentials: %w", err)
		}
		provider := credentials.CredentialsProviderFunc(func(ctx context.Context) (credentials.Credentials, error) {
			cred, err := cred.GetCredential()
			if err != nil {
				return credentials.Credentials{}, err
			}
			return credentials.Credentials{
				AccessKeyID:     *cred.AccessKeyId,
				AccessKeySecret: *cred.AccessKeySecret,
				SecurityToken:   *cred.SecurityToken,
			}, nil
		})

		cfg := oss.LoadDefaultConfig().WithRegion(region).WithCredentialsProvider(provider)
		return storage.NewAliyunOSS(oss.NewClient(cfg)), nil
	default:
		return nil, fmt.Errorf("unsupported cloud provider: %s", cfg.Provider)
	}
}

type Config struct {
	Tenants map[string]map[string]TenantConf `json:"tenants"`
}

type TenantConf struct {
	Frequency  string `json:"frequency"`
	TaskType   string `json:"taskType"`
	ProjectID  string `json:"projectId"`
	DataFormat string `json:"dataFormat"`
	DeployType string `json:"deployType"`
}

const (
	deployTypeServerless = "serverless"
	deployTypeDedicated  = "dedicated"
	deployTypePremium    = "premium"

	clusterIDAll = "all"

	logKindSlowlog    = "slowlog"
	logKindStatement  = "statement"
	taskTypeSlowQuery = "slowquery"
	taskTypeSQLStmt   = "sqlstatement"
)

func parseDeployType(deployType string) (string, error) {
	deployType = strings.ToLower(strings.TrimSpace(deployType))
	if deployType == "" {
		return deployTypeServerless, nil
	}
	switch deployType {
	case deployTypeServerless, deployTypeDedicated, deployTypePremium:
		return deployType, nil
	default:
		return "", fmt.Errorf("invalid deployType %q, must be serverless, dedicated, or premium", deployType)
	}
}

func parseLogKind(taskType string) (string, error) {
	switch strings.ToLower(strings.TrimSpace(taskType)) {
	case "", taskTypeSlowQuery, logKindSlowlog, "slow_query_etl":
		return logKindSlowlog, nil
	case taskTypeSQLStmt, logKindStatement:
		return logKindStatement, nil
	default:
		return "", fmt.Errorf("invalid taskType %q, must be slowquery or sqlstatement", taskType)
	}
}

// buildAllInOneSrcPrefix converts a normal serverless srcPrefix into the shared
// all-in-one prefix, e.g. 0/prod-devtier/serverless -> 0/prod-devtier/serverless-all-in-one.
func buildAllInOneSrcPrefix(prefix string) string {
	prefix = strings.TrimSuffix(prefix, "/")
	if strings.HasSuffix(prefix, "serverless-all-in-one") {
		return prefix
	}
	if strings.HasSuffix(prefix, "serverless") {
		return strings.TrimSuffix(prefix, "serverless") + "serverless-all-in-one"
	}
	return filepath.Join(prefix, "serverless-all-in-one")
}

func main() {
	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()
	log.Info("data_process is running")
	if err := run(ctx); err != nil {
		log.Panic("run failed", zap.Error(err))
	}
}

func GetConfigData() (*Config, error) {
	if len(confPath) == 0 {
		return nil, errors.New("config file path is empty")
	}
	fileContent, err := os.ReadFile(confPath)
	if err != nil {
		log.Fatal("failed to read config file", zap.Error(err))
		return nil, err
	}

	var config Config
	err = json.Unmarshal(fileContent, &config)
	if err != nil {
		log.Fatal("failed to parse json config json file", zap.Error(err))
		return nil, err
	}

	return &config, nil
}

func run(ctx context.Context) error {
	configInfo, err := GetConfigData()
	if err != nil {
		log.Fatal("Error getting ConfigMap: %v")
		return err
	}
	err = runAllClusterTasks(ctx, configInfo)
	if err != nil {
		return err
	}
	return nil
}

func runAllClusterTasks(ctx context.Context, cfg *Config) error {
	taskConfigs, err := createTaskConfigsFromConfigMap(cfg)
	if err != nil {
		return fmt.Errorf("failed to create task configs: %w", err)
	}

	var wg sync.WaitGroup
	errChan := make(chan error, len(taskConfigs))

	for _, innerConf := range taskConfigs {
		wg.Add(1)
		go func(c *ClusterTaskConfig) {
			defer wg.Done()
			if err := processClusterLevelTask(ctx, c); err != nil {
				log.Error("Task failed for tenant cluster",
					zap.String("tenantid", c.TenantID),
					zap.String("clusterid", c.ClusterID),
					zap.Error(err))
				errChan <- err
			}
		}(innerConf)
	}

	go func() {
		wg.Wait()
		close(errChan)
	}()

	return <-errChan
}

func createTaskConfigsFromConfigMap(cfg *Config) ([]*ClusterTaskConfig, error) {
	var configs []*ClusterTaskConfig

	for tenantID, tenant := range cfg.Tenants {
		for clusterID, cluster := range tenant {
			deployType, err := parseDeployType(cluster.DeployType)
			if err != nil {
				return nil, fmt.Errorf("tenant %s cluster %s: %w", tenantID, clusterID, err)
			}
			logKind, err := parseLogKind(cluster.TaskType)
			if err != nil {
				return nil, fmt.Errorf("tenant %s cluster %s: %w", tenantID, clusterID, err)
			}
			if clusterID == clusterIDAll && deployType != deployTypeServerless {
				return nil, fmt.Errorf("tenant %s cluster %s: cluster \"all\" is only supported for deployType serverless", tenantID, clusterID)
			}
			filterTenantID := ""
			if clusterID == clusterIDAll {
				filterTenantID = tenantID
			}
			singleConfig := &ClusterTaskConfig{
				Provider:       vendor,
				Region:         region,
				SourceBucket:   srcBucket,
				TargetBucket:   dstBucket,
				TenantID:       tenantID,
				ProjectID:      cluster.ProjectID,
				DataFormat:     cluster.DataFormat,
				ClusterID:      clusterID,
				TaskType:       cluster.TaskType,
				Frequency:      cluster.Frequency,
				DeployType:     deployType,
				LogKind:        logKind,
				FilterTenantID: filterTenantID,
			}
			configs = append(configs, singleConfig)
		}
	}

	return configs, nil
}

type ClusterTaskConfig struct {
	Provider       string
	Region         string
	SourceBucket   string
	TargetBucket   string
	TenantID       string
	ClusterID      string
	TaskType       string
	Frequency      string
	ProjectID      string
	DataFormat     string
	DeployType     string
	LogKind        string // slowlog or statement
	FilterTenantID string // when set, keep only records whose ServerlessTenantID matches
}

type StoragePaths struct {
	TargetFilePath      string
	SourceLogPath       string
	SourceMarkPointPath string
	StatusDir           string
}

// TaskStatusManager manages task status in memory
type TaskStatusManager struct {
	mu     sync.RWMutex
	tasks  map[int]bool // taskID -> isDone
	active map[int]bool // taskID -> isActive
}

func NewTaskStatusManager() *TaskStatusManager {
	return &TaskStatusManager{
		tasks:  make(map[int]bool),
		active: make(map[int]bool),
	}
}

// CreateTasks creates task IDs and marks them as active
func (m *TaskStatusManager) CreateTasks(count int) []int {
	m.mu.Lock()
	defer m.mu.Unlock()

	var taskIDs []int
	for i := 1; i <= count; i++ {
		m.tasks[i] = false
		m.active[i] = true
		taskIDs = append(taskIDs, i)
	}
	return taskIDs
}

// MarkDone marks a task as done
func (m *TaskStatusManager) MarkDone(taskID int) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.tasks[taskID] = true
	m.active[taskID] = false
}

// IsAllDone checks if all active tasks are done
func (m *TaskStatusManager) IsAllDone() bool {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if len(m.active) == 0 {
		return true
	}

	for taskID := range m.active {
		if !m.tasks[taskID] {
			return false
		}
	}
	return true
}

// Clear clears all task statuses
func (m *TaskStatusManager) Clear() {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.tasks = make(map[int]bool)
	m.active = make(map[int]bool)
}

func processClusterLevelTask(ctx context.Context, config *ClusterTaskConfig) error {
	paths := buildStoragePaths(config)

	sourceStorage, targetStorage, err := initStorages(ctx, config)
	if err != nil {
		return fmt.Errorf("failed to initialize storages: %w", err)
	}

	duration, err := time.ParseDuration(config.Frequency)
	if err != nil {
		return fmt.Errorf("invalid frequency format '%s' for tenant %s cluster %s: %w",
			config.Frequency, config.TenantID, config.ClusterID, err)
	}

	g, ctx := errgroup.WithContext(ctx)

	// Note: statusDir is no longer needed as task status is managed in memory
	if err := ensureDirectoriesExist(ctx, sourceStorage, targetStorage, config.SourceBucket, config.TargetBucket, paths.SourceMarkPointPath, paths.TargetFilePath); err != nil {
		return fmt.Errorf("meta dir is not exist: sourceBucket=%s markPath=%s targetBucket=%s targetPath=%s: %w",
			config.SourceBucket, paths.SourceMarkPointPath, config.TargetBucket, paths.TargetFilePath, err)
	}

	log.Info("start to monitor and process and async clean task.",
		zap.String("tenantId", config.TenantID),
		zap.String("clusterId", config.ClusterID),
		zap.String("projectId", config.ProjectID),
		zap.String("deployType", config.DeployType),
		zap.String("taskType", config.TaskType),
		zap.String("logKind", config.LogKind),
		zap.String("filterTenantID", config.FilterTenantID),
		zap.String("sourceBucket", config.SourceBucket),
		zap.String("targetBucket", config.TargetBucket),
		zap.String("srcPrefix", srcPrefix),
		zap.String("shareDataPrefix", shareDataPrefix),
		zap.String("sourceLogPath", paths.SourceLogPath),
		zap.String("sourceMarkPointPath", paths.SourceMarkPointPath),
		zap.String("targetFilePath", paths.TargetFilePath),
	)

	// Create task status manager for in-memory task tracking
	taskStatusMgr := NewTaskStatusManager()

	g.Go(func() error {
		return monitorAndProcess(
			ctx,
			sourceStorage,
			targetStorage,
			config,
			taskStatusMgr,
			paths.SourceMarkPointPath,
			paths.SourceLogPath,
			paths.TargetFilePath,
			duration,
		)
	})

	g.Go(func() error {
		cleanupTask, err := NewOSSCleanupTask(
			ctx,
			targetStorage,
			config.TargetBucket,
			paths.TargetFilePath,
			7,
		)
		if err != nil {
			return fmt.Errorf("task init failed: %w", err)
		}
		return cleanupTask.Run()
	})

	return g.Wait()
}

func buildStoragePaths(config *ClusterTaskConfig) *StoragePaths {
	logKind := config.LogKind
	if logKind == "" {
		logKind = logKindSlowlog
	}

	metaParts := []string{shareDataPrefix, "meta", config.TenantID}
	if config.ClusterID != clusterIDAll && config.ProjectID != "" {
		metaParts = append(metaParts, config.ProjectID)
	}
	metaParts = append(metaParts, config.ClusterID, "logs", logKind)

	targetParts := []string{"log_data", "data", config.TenantID}
	if config.ClusterID != clusterIDAll && config.ProjectID != "" {
		targetParts = append(targetParts, config.ProjectID)
	}
	targetParts = append(targetParts, config.ClusterID, logKind)

	return &StoragePaths{
		TargetFilePath:      filepath.Join(targetParts...) + "/",
		SourceLogPath:       buildSourceLogPath(config),
		SourceMarkPointPath: filepath.Join(append(append([]string{}, metaParts...), "pos")...) + "/",
		StatusDir:           filepath.Join(append(append([]string{}, metaParts...), "task_status")...) + "/",
	}
}

func buildSourceLogPath(config *ClusterTaskConfig) string {
	logKind := config.LogKind
	if logKind == "" {
		logKind = logKindSlowlog
	}
	if config.ClusterID == clusterIDAll {
		return filepath.Join(buildAllInOneSrcPrefix(srcPrefix), "logs", logKind) + "/"
	}
	if config.DeployType == deployTypePremium {
		return filepath.Join(srcPrefix, config.ClusterID, "logs", logKind) + "/"
	}
	return filepath.Join(
		srcPrefix,
		config.TenantID,
		config.ProjectID,
		config.ClusterID,
		"logs",
		logKind,
	) + "/"
}

func initStorages(ctx context.Context, config *ClusterTaskConfig) (storage.ListStorage, storage.ListStorage, error) {
	sourceObjectConfig := &ObjectConfig{
		Provider:   config.Provider,
		Region:     config.Region,
		BucketName: config.SourceBucket,
	}

	sourceStorage, err := NewObjectStorage(ctx, sourceObjectConfig)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create source storage client: %w", err)
	}
	// source and target object storage are beyond the same cloud and region
	targetObjectConfig := &ObjectConfig{
		Provider:   config.Provider,
		Region:     config.Region,
		BucketName: config.TargetBucket,
	}

	targetStorage, err := NewObjectStorage(ctx, targetObjectConfig)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create target storage client: %w", err)
	}

	return sourceStorage, targetStorage, nil
}

func monitorAndProcess(
	ctx context.Context,
	sourceStorage storage.ListStorage,
	targetStorage storage.ListStorage,
	config *ClusterTaskConfig,
	taskStatusMgr *TaskStatusManager,
	sourceMarkPointPath string,
	sourceLogPath string,
	targetFilePath string,
	frequency time.Duration,
) error {
	// Initial processing
	if err := processFilesIfNeeded(ctx, sourceStorage, targetStorage, config, taskStatusMgr, sourceMarkPointPath, sourceLogPath, targetFilePath); err != nil {
		return fmt.Errorf("initial process files failed: %w", err)
	}

	ticker := time.NewTicker(frequency)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			fmt.Println("stop monitor task")
			return nil
		case <-ticker.C:
			if taskStatusMgr.IsAllDone() {
				if err := processFilesIfNeeded(ctx, sourceStorage, targetStorage, config, taskStatusMgr, sourceMarkPointPath, sourceLogPath, targetFilePath); err != nil {
					return fmt.Errorf("process files failed: %w", err)
				}
			} else {
				fmt.Println("find unfinished tasks, wait for next round")
			}
		}
	}
}

// processFilesIfNeeded checks if there are new files to process before actually processing
func processFilesIfNeeded(
	ctx context.Context,
	sourceStorage storage.ListStorage,
	targetStorage storage.ListStorage,
	config *ClusterTaskConfig,
	taskStatusMgr *TaskStatusManager,
	sourceMarkPointPath string,
	sourceLogPath string,
	targetFilePath string,
) error {
	// Check if there are new files to process
	hasNewFiles, err := checkForNewFiles(ctx, sourceStorage, config.SourceBucket, sourceMarkPointPath, sourceLogPath)
	if err != nil {
		return fmt.Errorf("failed to check for new files: %w", err)
	}

	if !hasNewFiles {
		log.Info("no new files to process, skipping",
			zap.String("sourceBucket", config.SourceBucket),
			zap.String("sourceMarkPointPath", sourceMarkPointPath),
			zap.String("sourceLogPath", sourceLogPath),
		)
		return nil
	}

	log.Info("found new files to process, starting processing")
	return processFiles(ctx, sourceStorage, targetStorage, config, taskStatusMgr, sourceMarkPointPath, sourceLogPath, targetFilePath)
}

// checkForNewFiles checks if there are any files newer than the current marker.
// When a checkpoint exists, it lists incrementally from the marker key instead of scanning the whole prefix.
func checkForNewFiles(ctx context.Context, store storage.ListStorage, sourceBucket string, sourceMarkPointPath string, sourceLogPath string) (bool, error) {
	log.Info("checking for new files",
		zap.String("sourceBucket", sourceBucket),
		zap.String("sourceMarkPointPath", sourceMarkPointPath),
		zap.String("sourceLogPath", sourceLogPath),
	)

	objects, err := store.Objects(ctx, sourceBucket, sourceMarkPointPath)
	if err != nil {
		return false, err
	}

	newObjects := filterObjects(objects, sourceMarkPointPath)
	var markerFile string
	if len(newObjects) == 1 {
		markerFile = newObjects[0].Key
	}

	log.Info("marker checkpoint status",
		zap.Int("markerObjectsFound", len(newObjects)),
		zap.String("markerFile", markerFile),
	)

	baseTimestamp, startKey, inclusiveStart, err := getListingStart(sourceLogPath, markerFile)
	if err != nil {
		return false, err
	}
	scanned := 0
	found := false
	var sampleKeys []string
	err = store.ObjectsCallbackWithStartKey(ctx, sourceBucket, sourceLogPath, startKey, func(obj storage.ObjectMeta) bool {
		scanned++
		if len(sampleKeys) < 5 {
			sampleKeys = append(sampleKeys, obj.Key)
		}
		if strings.HasSuffix(obj.Key, "/") || (!inclusiveStart && obj.Key == startKey) {
			return true
		}
		ts, e := extractTimestamp(obj.Key)
		if e != nil {
			return true
		}
		if ts > baseTimestamp || (inclusiveStart && ts == baseTimestamp) {
			found = true
			return false
		}
		return true
	})
	log.Info("incremental file check completed",
		zap.String("startKey", startKey),
		zap.Int64("markerTimestamp", baseTimestamp),
		zap.Bool("inclusiveStart", inclusiveStart),
		zap.Int("objectsScanned", scanned),
		zap.Strings("sampleKeys", sampleKeys),
		zap.Bool("foundNewer", found),
	)
	if err != nil {
		return false, fmt.Errorf("list from checkpoint: %w", err)
	}
	return found, nil
}

func ensureDirectoriesExist(
	ctx context.Context,
	sourceStorage storage.ListStorage,
	targetStorage storage.ListStorage,
	sourceBucket string,
	targetBucket string,
	sourceMarkPointPath string,
	targetFilePath string,
) error {
	// Note: statusDir is no longer needed as task status is managed in memory

	if exists, err := sourceStorage.ObjectExists(ctx, sourceBucket, sourceMarkPointPath); err != nil {
		return fmt.Errorf("failed to check sourceMarkPointPath directory existence: %v", err)
	} else if !exists {
		emptyReader := bytes.NewReader([]byte{})
		if err := sourceStorage.Mkdir(ctx, sourceBucket, sourceMarkPointPath, emptyReader); err != nil {
			return fmt.Errorf("failed to create sourceMarkPointPath directory: %v", err)
		}
	}

	if exists, err := targetStorage.ObjectExists(ctx, targetBucket, targetFilePath); err != nil {
		return fmt.Errorf("failed to check targetFilePath directory existence: %v", err)
	} else if !exists {
		emptyReader := bytes.NewReader([]byte{})
		if err := targetStorage.Mkdir(ctx, targetBucket, targetFilePath, emptyReader); err != nil {
			return fmt.Errorf("failed to create targetFilePath directory: %v", err)
		}
	}

	return nil
}

func filterObjects(objects []storage.ObjectMeta, statusDir string) []storage.ObjectMeta {
	var filtered []storage.ObjectMeta
	for _, obj := range objects {
		if obj.Key != statusDir {
			filtered = append(filtered, obj)
		}
	}
	return filtered
}

// checkAllDone is no longer needed - task status is now managed in memory
// This function is kept for reference but not used

// processOneBatch processes one batch: split into groups, run processTask concurrently, merge and upload, then update checkpoint.
func processOneBatch(
	ctx context.Context,
	sourceStorage storage.ListStorage,
	targetStorage storage.ListStorage,
	config *ClusterTaskConfig,
	taskStatusMgr *TaskStatusManager,
	sourceMarkPointPath string,
	targetFilePath string,
	batch []string,
	oldMarkerKey string,
) (newMarkerKey string, err error) {
	if len(batch) == 0 {
		return oldMarkerKey, nil
	}
	taskCount := parallelWorkers
	if taskCount <= 0 {
		taskCount = defaultParallelWorkers
	}
	if len(batch) < taskCount {
		taskCount = len(batch)
	}
	taskStatusMgr.Clear()
	taskIDs := taskStatusMgr.CreateTasks(taskCount)
	if len(taskIDs) == 0 {
		return "", fmt.Errorf("can not create task statuses")
	}
	fileGroups := splitFiles(batch, len(taskIDs))
	filesPerWorker := make([]int, len(fileGroups))
	for i, group := range fileGroups {
		filesPerWorker[i] = len(group)
	}
	log.Info("batch processing started",
		zap.Int("totalFiles", len(batch)),
		zap.Int("workers", len(fileGroups)),
		zap.Ints("filesPerWorker", filesPerWorker),
		zap.String("firstFile", filepath.Base(batch[0])),
		zap.String("lastFile", filepath.Base(batch[len(batch)-1])),
	)
	namePrefix := config.LogKind
	if namePrefix == "" {
		namePrefix = logKindSlowlog
	}
	roller := newRollingGzipUploader(ctx, targetStorage, config.TargetBucket, targetFilePath, maxOutputFileSize, namePrefix)
	progress := newBatchProgressTracker(len(batch))
	roller.setProgressTracker(progress)
	batchStarted := time.Now()
	var wg sync.WaitGroup
	var taskErrors []error
	var mu sync.Mutex
	for i, taskID := range taskIDs {
		wg.Add(1)
		go func(id int, files []string) {
			defer wg.Done()
			err := processTask(ctx, sourceStorage, targetStorage, config, taskStatusMgr, id, files, roller, progress)
			if err != nil {
				mu.Lock()
				taskErrors = append(taskErrors, err)
				mu.Unlock()
			}
		}(taskID, fileGroups[i])
	}
	wg.Wait()
	if len(taskErrors) > 0 {
		for _, e := range taskErrors {
			log.Error("fail to execute task", zap.Error(e))
		}
		return "", fmt.Errorf("some task fail to execute")
	}
	if err := roller.Finish(); err != nil {
		return "", fmt.Errorf("failed to finalize rolling upload: %w", err)
	}
	log.Info("finished batch upload",
		zap.Int("files", len(batch)),
		zap.Int64("maxOutputFileSize", maxOutputFileSize),
		zap.Duration("elapsed", time.Since(batchStarted)),
		zap.Int64("downloadedBytes", progress.downloadedBytes.Load()),
		zap.Int64("uploadedBytes", progress.uploadedBytes.Load()),
		zap.Int64("parsedRecords", progress.parsedRecords.Load()),
		zap.Int64("uploadedRecords", progress.uploadedRecords.Load()),
		zap.Int64("uploadParts", progress.uploadParts.Load()),
	)
	lastProcessedFile := fileGroups[len(fileGroups)-1][len(fileGroups[len(fileGroups)-1])-1]
	newMarkerKey = filepath.Join(sourceMarkPointPath, filepath.Base(lastProcessedFile))
	if oldMarkerKey != "" {
		if err := sourceStorage.DeleteObject(ctx, config.SourceBucket, oldMarkerKey); err != nil {
			log.Error("fail to delete old marker", zap.String("key", oldMarkerKey), zap.Error(err))
		}
	}
	emptyReader := bytes.NewReader([]byte{})
	if err := sourceStorage.SaveObject(ctx, config.SourceBucket, newMarkerKey, emptyReader); err != nil {
		return "", fmt.Errorf("fail to create new marker file: %w", err)
	}
	log.Info("marker updated", zap.String("newMarker", newMarkerKey))
	return newMarkerKey, nil
}

func processFiles(
	ctx context.Context,
	sourceStorage storage.ListStorage,
	targetStorage storage.ListStorage,
	config *ClusterTaskConfig,
	taskStatusMgr *TaskStatusManager,
	sourceMarkPointPath string,
	sourceLogPath string,
	targetFilePath string,
) error {
	objects, err := sourceStorage.Objects(ctx, config.SourceBucket, sourceMarkPointPath)
	if err != nil {
		log.Error("failed to list marker files", zap.Error(err))
		return err
	}
	newObjects := filterObjects(objects, sourceMarkPointPath)
	if len(newObjects) > 1 {
		return fmt.Errorf("error process for mark files")
	}
	var markerFile string
	if len(newObjects) == 1 {
		markerFile = newObjects[0].Key
	}
	batchSize := maxFilesPerBatch
	if batchSize <= 0 {
		batchSize = defaultMaxFilesPerBatch
	}
	type fileWithTime struct {
		Key       string
		Timestamp int64
	}
	var batch []fileWithTime
	currentMarkerKey := markerFile
	baseTimestamp, startKey, inclusiveStart, err := getListingStart(sourceLogPath, markerFile)
	if err != nil {
		return err
	}
	flushBatch := func() error {
		if len(batch) == 0 {
			return nil
		}
		sort.Slice(batch, func(i, j int) bool { return batch[i].Timestamp < batch[j].Timestamp })
		keys := make([]string, len(batch))
		for i := range batch {
			keys[i] = batch[i].Key
		}
		newMarker, err := processOneBatch(ctx, sourceStorage, targetStorage, config, taskStatusMgr, sourceMarkPointPath, targetFilePath, keys, currentMarkerKey)
		if err != nil {
			return err
		}
		currentMarkerKey = newMarker
		batch = batch[:0]
		return nil
	}
	var flushErr error
	err = sourceStorage.ObjectsCallbackWithStartKey(ctx, config.SourceBucket, sourceLogPath, startKey, func(obj storage.ObjectMeta) bool {
		if strings.HasSuffix(obj.Key, "/") || (!inclusiveStart && obj.Key == startKey) {
			return true
		}
		ts, e := extractTimestamp(obj.Key)
		if e != nil {
			return true
		}
		if ts < baseTimestamp || (!inclusiveStart && ts == baseTimestamp) {
			return true
		}
		batch = append(batch, fileWithTime{Key: obj.Key, Timestamp: ts})
		if len(batch) >= batchSize {
			flushErr = flushBatch()
			if flushErr != nil {
				return false
			}
		}
		return true
	})
	if flushErr != nil {
		return flushErr
	}
	if err != nil {
		return fmt.Errorf("list from checkpoint: %w", err)
	}
	if err := flushBatch(); err != nil {
		return err
	}
	log.Info("successfully finished processing files")
	return nil
}

func getListingStart(sourceLogPath string, markerFile string) (baseTimestamp int64, startKey string, inclusiveStart bool, err error) {
	if markerFile != "" {
		baseTimestamp, err = extractTimestamp(markerFile)
		if err != nil {
			return 0, "", false, fmt.Errorf("fail to parse file ts: %w", err)
		}
		return baseTimestamp, strings.TrimSuffix(sourceLogPath, "/") + "/" + filepath.Base(markerFile), false, nil
	}

	prefix := initialListingPrefix()
	baseTimestamp = prefix * initialMarkerTrimDigits
	return baseTimestamp, fmt.Sprintf("%s/%d", strings.TrimSuffix(sourceLogPath, "/"), prefix), true, nil
}

func initialListingPrefix() int64 {
	if initialMarkerPrefix > 0 {
		return initialMarkerPrefix
	}
	now := timeNow().Unix()
	if now <= 0 {
		return 0
	}
	return now / initialMarkerTrimDigits
}

// createTaskStatusFiles is no longer needed - task status is now managed in memory
// This function is kept for reference but not used

func splitFiles(files []string, groups int) [][]string {
	if groups <= 0 || len(files) == 0 {
		return nil
	}

	if groups > len(files) {
		groups = len(files)
	}

	result := make([][]string, groups)
	perGroup := len(files) / groups
	remainder := len(files) % groups

	current := 0
	for i := 0; i < groups; i++ {
		groupSize := perGroup
		if i < remainder {
			groupSize++
		}

		end := current + groupSize
		result[i] = files[current:end]
		current = end
	}

	return result
}

type sourceFileStats struct {
	DownloadDuration time.Duration
	ProcessDuration  time.Duration
	DownloadBytes    int64
	Records          int
}

func processTask(
	ctx context.Context,
	sourceStorage storage.ListStorage,
	targetStorage storage.ListStorage,
	config *ClusterTaskConfig,
	taskStatusMgr *TaskStatusManager,
	id int,
	files []string,
	roller *rollingGzipUploader,
	progress *batchProgressTracker,
) error {
	if len(files) == 0 {
		taskStatusMgr.MarkDone(id)
		return nil
	}

	log.Info("worker started",
		zap.Int("taskId", id),
		zap.Int("assignedFiles", len(files)),
		zap.String("firstFile", filepath.Base(files[0])),
		zap.String("lastFile", filepath.Base(files[len(files)-1])),
	)

	task := &Task{
		ctx:            ctx,
		sourceBucket:   config.SourceBucket,
		targetBucket:   config.TargetBucket,
		tenantId:       config.TenantID,
		clusterId:      config.ClusterID,
		sourceStorage:  sourceStorage,
		targetStorage:  targetStorage,
		id:             id,
		filterTenantID: config.FilterTenantID,
		logKind:        config.LogKind,
	}

	workerStarted := time.Now()
	successFiles := 0
	var totalRecords int
	var totalDownloadBytes int64
	for i, file := range files {
		fileStarted := time.Now()
		stats, err := task.streamFileToRoller(ctx, file, roller)
		if err != nil {
			log.Error("failed to stream logs from file",
				zap.Int("taskId", id),
				zap.Int("fileProgress", i+1),
				zap.Int("taskFileTotal", len(files)),
				zap.String("file", filepath.Base(file)),
				zap.String("key", file),
				zap.Duration("elapsed", time.Since(fileStarted)),
				zap.Error(err),
			)
			processingErrors.Inc()
			continue
		}
		if err := roller.FlushPending(); err != nil {
			return fmt.Errorf("flush rolling uploader after file %s: %w", file, err)
		}
		successFiles++
		totalRecords += stats.Records
		totalDownloadBytes += stats.DownloadBytes
		progress.markFileProcessed(stats.DownloadBytes, stats.Records, filepath.Base(file))
	}

	taskStatusMgr.MarkDone(id)

	log.Info("worker finished",
		zap.Int("taskId", id),
		zap.Int("assignedFiles", len(files)),
		zap.Int("processedFiles", successFiles),
		zap.Int("records", totalRecords),
		zap.Int64("downloadedBytes", totalDownloadBytes),
		zap.Duration("elapsed", time.Since(workerStarted)),
	)
	return nil
}

func FilterNewerFiles(ctx context.Context, storage storage.ListStorage, sourceBucket string, sourceLogPath string, markerFile string) ([]string, error) {
	baseTimestamp := int64(-1)

	if len(markerFile) > 0 {
		ts, err := extractTimestamp(markerFile)
		if err != nil {
			return nil, fmt.Errorf("fail to parse file ts: %w", err)
		}
		baseTimestamp = ts
	}

	type fileWithTime struct {
		Key       string
		Timestamp int64
	}
	var files []fileWithTime

	marker := ""
	totalListed := 0
	for {
		lsRes, nextMarker, err := storage.ListObjectsPaged(ctx, sourceBucket, sourceLogPath, 1000, marker)
		if err != nil {
			return nil, fmt.Errorf("fail to list files: %w", err)
		}

		filterRes := filterObjects(lsRes, sourceLogPath)
		totalListed += len(filterRes)

		for _, obj := range filterRes {
			if strings.HasSuffix(obj.Key, "/") || obj.Key == markerFile {
				continue
			}

			timestamp, err := extractTimestamp(obj.Key)
			if err != nil {
				continue
			}

			if timestamp > baseTimestamp {
				files = append(files, fileWithTime{
					Key:       obj.Key,
					Timestamp: timestamp,
				})
			}
		}

		if nextMarker == "" {
			break
		}
		marker = nextMarker
	}

	// Only log summary, not each page
	log.Info("file listing completed",
		zap.Int("total_files_listed", totalListed),
		zap.Int("files_newer_than_marker", len(files)))

	sort.Slice(files, func(i, j int) bool {
		return files[i].Timestamp < files[j].Timestamp
	})

	newerFiles := make([]string, 0, len(files))
	for _, f := range files {
		newerFiles = append(newerFiles, f.Key)
	}

	log.Info(" find " + strconv.Itoa(len(newerFiles)) + " files after filter and sort.")
	return newerFiles, nil
}

func extractTimestamp(filename string) (int64, error) {
	base := filepath.Base(filename)
	parts := strings.Split(base, "-")
	if len(parts) < 2 {
		return 0, fmt.Errorf("invalide file format")
	}

	timestamp, err := strconv.ParseInt(parts[0], 10, 64)
	if err != nil {
		return 0, fmt.Errorf("fail to parse time stamp: %w", err)
	}

	return timestamp, nil
}

type OSSCleanupTask struct {
	ctx        context.Context
	Client     storage.ListStorage
	BucketName string
	Prefix     string
	Retention  int
	StopChan   chan struct{}
}

func NewOSSCleanupTask(ctx context.Context, storage storage.ListStorage, bucket string, prefix string, retention int) (*OSSCleanupTask, error) {
	return &OSSCleanupTask{
		ctx:        ctx,
		Client:     storage,
		BucketName: bucket,
		Prefix:     prefix,
		Retention:  retention,
		StopChan:   make(chan struct{}),
	}, nil
}

func (t *OSSCleanupTask) Run() error {
	ticker := time.NewTicker(24 * time.Hour)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			if err := t.cleanup(); err != nil {
				log.Info("clean failed : %v")
			}
		case <-t.StopChan:
			log.Info("stop OSS clean task ")
			return nil
		}
	}
}
func (t *OSSCleanupTask) cleanup() error {
	cutoff := time.Now().AddDate(0, 0, -t.Retention)
	nextToken := ""

	for {
		lsRes, token, err := t.Client.ListObjectsPaged(
			t.ctx,
			t.BucketName,
			t.Prefix,
			100,
			nextToken,
		)
		if err != nil {
			return fmt.Errorf("fail to list file: %w", err)
		}

		if len(lsRes) == 0 {
			fmt.Printf("to the end")
			return nil
		}

		deletedCount := 0
		for _, obj := range lsRes {
			if len(obj.Key) < len(t.Prefix)+11 {
				fmt.Printf("skip invalid file: %s", obj.Key)
				continue
			}

			timePart := (obj.Key)[len(t.Prefix) : len(t.Prefix)+11]
			fileTime, err := time.Parse("20060102_15", timePart)
			if err != nil {
				fmt.Printf("fail to parse file name: %s (%v)", obj.Key, err)
				continue
			}

			if fileTime.Before(cutoff) {
				if err := t.Client.DeleteObject(context.TODO(), t.BucketName, obj.Key); err != nil {
					fmt.Printf("delete fail: %s (%v)", obj.Key, err)
				} else {
					deletedCount++
					fmt.Printf("deleted: %s (time: %v)", obj.Key, fileTime.Format("2006-01-02 15:04"))
				}
			}
		}
		fmt.Printf("delete  %d objects", deletedCount)

		if token == "" {
			fmt.Printf("deal with all object")
			return nil
		}

		nextToken = token
		fmt.Printf("prepare next page,token: %s", nextToken)
	}
}

func (t *OSSCleanupTask) Stop() {
	close(t.StopChan)
}

type SlowLog struct {
	Time                      time.Time
	Instance                  string
	User                      string
	Host                      string
	ConnID                    int
	SessionAlias              string
	ExecRetryCount            int
	ExecRetryTime             float64
	QueryTime                 float64
	ParseTime                 float64
	CompileTime               float64
	RewriteTime               float64
	PreprocSubqueries         int
	PreprocSubqueriesTime     float64
	OptimizeTime              float64
	WaitTS                    float64
	PrewriteTime              float64
	WaitPrewriteBinlogTime    float64
	CommitTime                float64
	GetCommitTsTime           float64
	CommitBackoffTime         float64
	BackoffTypes              string
	ResolveLockTime           float64
	LocalLatchWaitTime        float64
	WriteKeys                 int
	WriteSize                 int
	PrewriteRegion            int
	TxnRetry                  int
	CopTime                   float64
	ProcessTime               float64
	WaitTime                  float64
	BackoffTime               float64
	LockKeysTime              float64
	RequestCount              int
	TotalKeys                 int
	ProcessKeys               int
	RocksdbDeleteSkippedCount int
	RocksdbKeySkippedCount    int
	RocksdbBlockCacheHitCount int
	RocksdbBlockReadCount     int
	RocksdbBlockReadByte      int
	DB                        string
	IndexNames                string
	IsInternal                bool
	Digest                    string
	Stats                     string
	CopProcAvg                float64
	CopProcP90                float64
	CopProcMax                float64
	CopProcAddr               string
	CopWaitAvg                float64
	CopWaitP90                float64
	CopWaitMax                float64
	CopWaitAddr               string
	MemMax                    int
	DiskMax                   int
	KVTotal                   float64
	PDTotal                   float64
	BackoffTotal              float64
	WriteSQLResponseTotal     float64
	ResultRows                int
	Warnings                  string
	BackoffDetail             string
	Prepared                  bool
	Succ                      bool
	IsExplicitTxn             bool
	IsWriteCacheTable         bool
	PlanFromCache             bool
	PlanFromBinding           bool
	HasMoreResults            bool
	ResourceGroup             string
	RequestUnitRead           float64
	RequestUnitWrite          float64
	TimeQueuedByRC            float64
	Plan                      string
	PlanDigest                string
	BinaryPlan                string
	PrevStmt                  string
	Query                     string
	ServerlessTenantID        string
	ServerlessProjectID       string
	ServerlessClusterID       string
	WorkerRole                string
	IsTiFlash                 bool
	TxnStartTs                uint64
}

type StreamParser struct {
	reader      *bufio.Reader
	src         io.Reader
	pendingLine string // first line of next record when we end current on "# Time"
}

func NewStreamParser(reader io.Reader) *StreamParser {
	return &StreamParser{
		reader: bufio.NewReader(reader),
		src:    reader,
	}
}

// isTimeLine reports whether the line starts a new slow log record (# Time: ...).
func isTimeLine(line string) bool {
	if line == "" {
		return false
	}
	return (line[0] == '2' && strings.Contains(line, "# Time: ")) ||
		strings.HasPrefix(line, "# Time: ")
}

func (p *StreamParser) ReadNext() (log *SlowLog, err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic while reading log: %v", r)
		}
	}()

	var lines []string
	if p.pendingLine != "" {
		lines = append(lines, p.pendingLine)
		p.pendingLine = ""
	}

	for {
		line, err := p.reader.ReadString('\n')
		if err != nil {
			if err == io.EOF {
				if len(line) > 0 {
					lineStr := strings.TrimSpace(line)
					if lineStr != "" {
						lines = append(lines, lineStr)
					}
				}
				if len(lines) > 0 {
					return safeParse(lines)
				}
				return nil, nil
			}
			return nil, fmt.Errorf("read failed: %w", err)
		}

		lineStr := strings.TrimSpace(line)
		if lineStr == "" {
			continue
		}

		// Next record starts with "# Time: " — current record ends at end of previous line (or EOF).
		if isTimeLine(lineStr) {
			if len(lines) > 0 {
				p.pendingLine = lineStr
				return safeParse(lines)
			}
		}

		lines = append(lines, lineStr)

		// Also end record when we see a SQL line (ends with ;, not comment/use) for backward compatibility.
		if strings.HasSuffix(lineStr, ";") && !isComment(lineStr) && !strings.HasPrefix(strings.ToLower(lineStr), "use ") {
			return safeParse(lines)
		}
	}
}

func isComment(line string) bool {
	return strings.HasPrefix(line, "#") ||
		strings.HasPrefix(line, "--") ||
		strings.HasPrefix(line, "/*")
}

func safeParse(lines []string) (_ *SlowLog, err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("parse panic: %v\ninput: %v", r, lines)
		}
	}()

	log, err := parse(lines)
	if err != nil {
		return nil, fmt.Errorf("parse error (lines=%d, first=%s): %w", len(lines), truncateForLog(lines[0], maxLogSnippetLen), err)
	}
	return log, nil
}

func parse(lines []string) (*SlowLog, error) {
	if len(lines) == 0 {
		return nil, nil
	}

	sl := &SlowLog{}
	for _, line := range lines {
		if err := parseLine(sl, line); err != nil {
			return nil, err
		}
	}
	return sl, nil
}

// // parseLine parses a single line in a slow log entry
// func parseLine(sl *SlowLog, line string) error {
// 	if line == "" {
// 		return nil
// 	}

// 	// Fast path for first line
// 	if (line[0] == '2' && strings.Contains(line, "# Time: ")) || strings.HasPrefix(line, "# Time: ") {
// 		arr := strings.SplitN(line, "# Time: ", 2)
// 		if len(arr) != 2 {
// 			return fmt.Errorf("invalid first line: %s", line)
// 		}

// 		if arr[0] != "" {
// 			m := vectorLogPrefixReg.FindStringSubmatch(arr[0])
// 			if len(m) > 0 {
// 				sl.Instance = m[0]
// 			}
// 		}

// 		if !strings.HasSuffix(arr[1], "Z") {
// 			return fmt.Errorf("unsupported timezone")
// 		}

// 		timeStr := strings.TrimSuffix(arr[1], "Z")
// 		if len(timeStr) > 26 {
// 			timeStr = timeStr[:26]
// 		}

// 		t, err := time.Parse("2006-01-02T15:04:05.999999", timeStr)
// 		if err != nil {
// 			return fmt.Errorf("failed to parse time: %v", err)
// 		}
// 		sl.Time = t.UTC()
// 	} else if strings.HasPrefix(line, "# ") {
// 		switch {
// 		case strings.HasPrefix(line, "# User@Host: "):
// 			arr := strings.SplitN(line[len("# User@Host: "):], " @ ", 2)
// 			if len(arr) != 2 {
// 				return fmt.Errorf("invalid user and host line: %s", line)
// 			}

// 			userArr := strings.Split(arr[0], "[")
// 			if len(userArr) < 2 {
// 				return fmt.Errorf("invalid user and host line: %s", line)
// 			}
// 			sl.User = strings.TrimSpace(userArr[0])

// 			hostArr := strings.Split(arr[1], "[")
// 			if len(hostArr) < 2 {
// 				return fmt.Errorf("invalid user and host line: %s", line)
// 			}
// 			sl.Host = strings.TrimSpace(hostArr[0])

// 		case strings.HasPrefix(line, "# Session_alias: "):
// 			sl.SessionAlias = strings.TrimSpace(line[len("# Session_alias: "):])

// 		case strings.HasPrefix(line, "# Backoff_types: "):
// 			sl.BackoffTypes = strings.TrimSpace(line[len("# Backoff_types: "):])

// 		case strings.HasPrefix(line, "# DB: "):
// 			sl.DB = strings.TrimSpace(line[len("# DB: "):])

// 		case strings.HasPrefix(line, "# Index_names: "):
// 			sl.IndexNames = strings.TrimSpace(line[len("# Index_names: "):])

// 		case strings.HasPrefix(line, "# Digest: "):
// 			sl.Digest = strings.TrimSpace(line[len("# Digest: "):])

// 		case strings.HasPrefix(line, "# Stats: "):
// 			sl.Stats = strings.TrimSpace(line[len("# Stats: "):])

// 		case strings.HasPrefix(line, "# Warnings: "):
// 			sl.Warnings = strings.TrimSpace(line[len("# Warnings: "):])

// 		case strings.HasPrefix(line, "# Cop_backoff_"):
// 			sl.BackoffDetail = strings.TrimSpace(strings.TrimPrefix(line, "# "))

// 		case strings.HasPrefix(line, "# Plan: "):
// 			sl.Plan = strings.TrimSpace(line[len("# Plan: "):])

// 		case strings.HasPrefix(line, "# Resource_group: "):
// 			sl.ResourceGroup = strings.TrimSpace(line[len("# Resource_group: "):])

// 		case strings.HasPrefix(line, "# Plan_digest: "):
// 			sl.PlanDigest = strings.TrimSpace(line[len("# Plan_digest: "):])

// 		case strings.HasPrefix(line, "# Binary_plan: "):
// 			sl.BinaryPlan = strings.TrimSpace(line[len("# Binary_plan: "):])

// 		case strings.HasPrefix(line, "# Prev_stmt: "):
// 			sl.PrevStmt = strings.TrimSpace(line[len("# Prev_stmt: "):])

// 		case strings.HasPrefix(line, "# Serverless_tenant_ID: "):
// 			sl.ServerlessTenantID = strings.TrimSpace(line[len("# Serverless_tenant_ID: "):])

// 		case strings.HasPrefix(line, "# Serverless_project_ID: "):
// 			sl.ServerlessProjectID = strings.TrimSpace(line[len("# Serverless_project_ID: "):])

// 		case strings.HasPrefix(line, "# Serverless_cluster_ID: "):
// 			sl.ServerlessClusterID = strings.TrimSpace(line[len("# Serverless_cluster_ID: "):])

// 		case strings.HasPrefix(line, "# Worker_role: "):
// 			sl.WorkerRole = strings.TrimSpace(line[len("# Worker_role: "):])

// 		default:
// 			matches := splitByColonReg.FindAllStringSubmatch(strings.TrimPrefix(line, "# "), -1)
// 			for _, match := range matches {
// 				if len(match) != 3 {
// 					continue
// 				}
// 				k := match[1]
// 				v := match[2]

// 				switch k {
// 				case "Txn_start_ts":
// 					val, err := strconv.ParseInt(v, 10, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Txn_start_ts: %v", err)
// 					}
// 					sl.TxnStartTs = val
// 				case "Conn_ID":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Conn_ID: %v", err)
// 					}
// 					sl.ConnID = val
// 				case "Exec_retry_count":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Exec_retry_count: %v", err)
// 					}
// 					sl.ExecRetryCount = val
// 				case "Exec_retry_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Exec_retry_time: %v", err)
// 					}
// 					sl.ExecRetryTime = val
// 				case "Query_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Query_time: %v", err)
// 					}
// 					sl.QueryTime = val
// 				case "Parse_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Parse_time: %v", err)
// 					}
// 					sl.ParseTime = val
// 				case "Compile_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Compile_time: %v", err)
// 					}
// 					sl.CompileTime = val
// 				case "Rewrite_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Rewrite_time: %v", err)
// 					}
// 					sl.RewriteTime = val
// 				case "Preproc_subqueries":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Preproc_subqueries: %v", err)
// 					}
// 					sl.PreprocSubqueries = val
// 				case "Preproc_subqueries_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Preproc_subqueries_time: %v", err)
// 					}
// 					sl.PreprocSubqueriesTime = val
// 				case "Optimize_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Optimize_time: %v", err)
// 					}
// 					sl.OptimizeTime = val
// 				case "Wait_TS":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Wait_TS: %v", err)
// 					}
// 					sl.WaitTS = val
// 				case "Prewrite_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Prewrite_time: %v", err)
// 					}
// 					sl.PrewriteTime = val
// 				case "Wait_prewrite_binlog_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Wait_prewrite_binlog_time: %v", err)
// 					}
// 					sl.WaitPrewriteBinlogTime = val
// 				case "Commit_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Commit_time: %v", err)
// 					}
// 					sl.CommitTime = val
// 				case "Get_commit_ts_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Get_commit_ts_time: %v", err)
// 					}
// 					sl.GetCommitTsTime = val
// 				case "Commit_backoff_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Commit_backoff_time: %v", err)
// 					}
// 					sl.CommitBackoffTime = val
// 				case "Resolve_lock_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Resolve_lock_time: %v", err)
// 					}
// 					sl.ResolveLockTime = val
// 				case "Local_latch_wait_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Local_latch_wait_time: %v", err)
// 					}
// 					sl.LocalLatchWaitTime = val
// 				case "Write_keys":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Write_keys: %v", err)
// 					}
// 					sl.WriteKeys = val
// 				case "Write_size":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Write_size: %v", err)
// 					}
// 					sl.WriteSize = val
// 				case "Prewrite_region":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Prewrite_region: %v", err)
// 					}
// 					sl.PrewriteRegion = val
// 				case "Txn_retry":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Txn_retry: %v", err)
// 					}
// 					sl.TxnRetry = val
// 				case "Cop_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_time: %v", err)
// 					}
// 					sl.CopTime = val
// 				case "Process_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Process_time: %v", err)
// 					}
// 					sl.ProcessTime = val
// 				case "Wait_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Wait_time: %v", err)
// 					}
// 					sl.WaitTime = val
// 				case "Backoff_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Backoff_time: %v", err)
// 					}
// 					sl.BackoffTime = val
// 				case "LockKeys_time":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid LockKeys_time: %v", err)
// 					}
// 					sl.LockKeysTime = val
// 				case "Request_count":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Request_count: %v", err)
// 					}
// 					sl.RequestCount = val
// 				case "Total_keys":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Total_keys: %v", err)
// 					}
// 					sl.TotalKeys = val
// 				case "Process_keys":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Process_keys: %v", err)
// 					}
// 					sl.ProcessKeys = val
// 				case "Rocksdb_delete_skipped_count":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Rocksdb_delete_skipped_count: %v", err)
// 					}
// 					sl.RocksdbDeleteSkippedCount = val
// 				case "Rocksdb_key_skipped_count":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Rocksdb_key_skipped_count: %v", err)
// 					}
// 					sl.RocksdbKeySkippedCount = val
// 				case "Rocksdb_block_cache_hit_count":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Rocksdb_block_cache_hit_count: %v", err)
// 					}
// 					sl.RocksdbBlockCacheHitCount = val
// 				case "Rocksdb_block_read_count":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Rocksdb_block_read_count: %v", err)
// 					}
// 					sl.RocksdbBlockReadCount = val
// 				case "Rocksdb_block_read_byte":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Rocksdb_block_read_byte: %v", err)
// 					}
// 					sl.RocksdbBlockReadByte = val
// 				case "Is_internal":
// 					sl.IsInternal = strToBool(v)
// 				case "Cop_proc_avg":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_proc_avg: %v", err)
// 					}
// 					sl.CopProcAvg = val
// 				case "Cop_proc_p90":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_proc_p90: %v", err)
// 					}
// 					sl.CopProcP90 = val
// 				case "Cop_proc_max":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_proc_max: %v", err)
// 					}
// 					sl.CopProcMax = val
// 				case "Cop_proc_addr":
// 					sl.CopProcAddr = v
// 				case "Cop_wait_avg":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_wait_avg: %v", err)
// 					}
// 					sl.CopWaitAvg = val
// 				case "Cop_wait_p90":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_wait_p90: %v", err)
// 					}
// 					sl.CopWaitP90 = val
// 				case "Cop_wait_max":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Cop_wait_max: %v", err)
// 					}
// 					sl.CopWaitMax = val
// 				case "Cop_wait_addr":
// 					sl.CopWaitAddr = v
// 				case "Mem_max":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Mem_max: %v", err)
// 					}
// 					sl.MemMax = val
// 				case "Disk_max":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Disk_max: %v", err)
// 					}
// 					sl.DiskMax = val
// 				case "KV_total":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid KV_total: %v", err)
// 					}
// 					sl.KVTotal = val
// 				case "PD_total":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid PD_total: %v", err)
// 					}
// 					sl.PDTotal = val
// 				case "Backoff_total":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Backoff_total: %v", err)
// 					}
// 					sl.BackoffTotal = val
// 				case "Write_sql_response_total":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Write_sql_response_total: %v", err)
// 					}
// 					sl.WriteSQLResponseTotal = val
// 				case "Result_rows":
// 					val, err := strconv.Atoi(v)
// 					if err != nil {
// 						return fmt.Errorf("invalid Result_rows: %v", err)
// 					}
// 					sl.ResultRows = val
// 				case "Prepared":
// 					sl.Prepared = strToBool(v)
// 				case "Succ":
// 					sl.Succ = strToBool(v)
// 				case "IsExplicitTxn":
// 					sl.IsExplicitTxn = strToBool(v)
// 				case "IsWriteCacheTable":
// 					sl.IsWriteCacheTable = strToBool(v)
// 				case "Plan_from_cache":
// 					sl.PlanFromCache = strToBool(v)
// 				case "Plan_from_binding":
// 					sl.PlanFromBinding = strToBool(v)
// 				case "Has_more_results":
// 					sl.HasMoreResults = strToBool(v)
// 				case "Request_unit_read":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Request_unit_read: %v", err)
// 					}
// 					sl.RequestUnitRead = val
// 				case "Request_unit_write":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Request_unit_write: %v", err)
// 					}
// 					sl.RequestUnitWrite = val
// 				case "Time_queued_by_rc":
// 					val, err := strconv.ParseFloat(v, 64)
// 					if err != nil {
// 						return fmt.Errorf("invalid Time_queued_by_rc: %v", err)
// 					}
// 					sl.TimeQueuedByRC = val
// 				case "Is_tiflash":
// 					sl.IsTiFlash = strToBool(v)
// 				}
// 			}
// 		}
// 	} else if strings.HasSuffix(line, ";") {
// 		if strings.HasPrefix(line, "use") {
// 			// Ignore 'use DB' statements
// 			return nil
// 		}
// 		sl.Query = line
// 	} else {
// 		if strings.Contains(line, "following end of new file") {
// 			return nil
// 		}
// 		return fmt.Errorf("invalid line: %s", line)
// 	}

// 	return nil
// }

type lineHandler interface {
	canHandle(line string) bool
	handle(sl *SlowLog, line string) error
}

type timeLineHandler struct{}

func (h *timeLineHandler) canHandle(line string) bool {
	return (line[0] == '2' && strings.Contains(line, "# Time: ")) ||
		strings.HasPrefix(line, "# Time: ")
}

func (h *timeLineHandler) handle(sl *SlowLog, line string) error {
	arr := strings.SplitN(line, "# Time: ", 2)
	if len(arr) != 2 {
		return fmt.Errorf("invalid first line: %s", line)
	}

	if arr[0] != "" {
		m := vectorLogPrefixReg.FindStringSubmatch(arr[0])
		if len(m) > 0 {
			sl.Instance = m[0]
		}
	}

	if !strings.HasSuffix(arr[1], "Z") {
		return fmt.Errorf("unsupported timezone")
	}

	timeStr := strings.TrimSuffix(arr[1], "Z")
	if len(timeStr) > 26 {
		timeStr = timeStr[:26]
	}

	t, err := time.Parse("2006-01-02T15:04:05.999999", timeStr)
	if err != nil {
		return fmt.Errorf("failed to parse time: %v", err)
	}
	sl.Time = t.UTC()
	return nil
}

type simpleKVHandler struct {
	prefix    string
	fieldFunc func(sl *SlowLog, value string)
}

func (h *simpleKVHandler) canHandle(line string) bool {
	return strings.HasPrefix(line, h.prefix)
}

func (h *simpleKVHandler) handle(sl *SlowLog, line string) error {
	value := strings.TrimSpace(line[len(h.prefix):])
	h.fieldFunc(sl, value)
	return nil
}

type userHostHandler struct{}

func (h *userHostHandler) canHandle(line string) bool {
	return strings.HasPrefix(line, "# User@Host: ")
}

func (h *userHostHandler) handle(sl *SlowLog, line string) error {
	arr := strings.SplitN(line[len("# User@Host: "):], " @ ", 2)
	if len(arr) != 2 {
		return fmt.Errorf("invalid user and host line: %s", line)
	}

	userArr := strings.Split(arr[0], "[")
	if len(userArr) < 2 {
		return fmt.Errorf("invalid user and host line: %s", line)
	}
	sl.User = strings.TrimSpace(userArr[0])

	hostArr := strings.Split(arr[1], "[")
	if len(hostArr) < 2 {
		return fmt.Errorf("invalid user and host line: %s", line)
	}
	sl.Host = strings.TrimSpace(hostArr[0])
	return nil
}

type complexKVHandler struct{}

func (h *complexKVHandler) canHandle(line string) bool {
	return strings.HasPrefix(line, "# ") && strings.Contains(line, ":")
}

func (h *complexKVHandler) handle(sl *SlowLog, line string) error {
	matches := splitByColonReg.FindAllStringSubmatch(strings.TrimPrefix(line, "# "), -1)
	for _, match := range matches {
		if len(match) != 3 {
			continue
		}
		k := match[1]
		v := match[2]

		if parser, exists := fieldParsers[k]; exists {
			if err := parser(sl, v); err != nil {
				return err
			}
		}
	}
	return nil
}

type queryHandler struct{}

func (h *queryHandler) canHandle(line string) bool {
	return strings.HasSuffix(line, ";")
}

func (h *queryHandler) handle(sl *SlowLog, line string) error {
	if strings.HasPrefix(line, "use") {
		// Ignore 'use DB' statements
		return nil
	}
	sl.Query = line
	return nil
}

type ignoreHandler struct{}

func (h *ignoreHandler) canHandle(line string) bool {
	return strings.Contains(line, "following end of new file")
}

func (h *ignoreHandler) handle(sl *SlowLog, line string) error {
	return nil
}

type fieldParser func(sl *SlowLog, value string) error

var fieldParsers = map[string]fieldParser{
	"Txn_start_ts": func(sl *SlowLog, v string) error {
		val, err := strconv.ParseUint(v, 10, 64)
		if err != nil {
			return fmt.Errorf("invalid Txn_start_ts: %v", err)
		}
		sl.TxnStartTs = val
		return nil
	},
	"Conn_ID":                   intParser(func(sl *SlowLog, val int) { sl.ConnID = val }),
	"Exec_retry_count":          intParser(func(sl *SlowLog, val int) { sl.ExecRetryCount = val }),
	"Exec_retry_time":           floatParser(func(sl *SlowLog, val float64) { sl.ExecRetryTime = val }),
	"Query_time":                floatParser(func(sl *SlowLog, val float64) { sl.QueryTime = val }),
	"Parse_time":                floatParser(func(sl *SlowLog, val float64) { sl.ParseTime = val }),
	"Compile_time":              floatParser(func(sl *SlowLog, val float64) { sl.CompileTime = val }),
	"Rewrite_time":              floatParser(func(sl *SlowLog, val float64) { sl.RewriteTime = val }),
	"Preproc_subqueries":        intParser(func(sl *SlowLog, val int) { sl.PreprocSubqueries = val }),
	"Preproc_subqueries_time":   floatParser(func(sl *SlowLog, val float64) { sl.PreprocSubqueriesTime = val }),
	"Optimize_time":             floatParser(func(sl *SlowLog, val float64) { sl.OptimizeTime = val }),
	"Wait_TS":                   floatParser(func(sl *SlowLog, val float64) { sl.WaitTS = val }),
	"Prewrite_time":             floatParser(func(sl *SlowLog, val float64) { sl.PrewriteTime = val }),
	"Wait_prewrite_binlog_time": floatParser(func(sl *SlowLog, val float64) { sl.WaitPrewriteBinlogTime = val }),
	"Commit_time":               floatParser(func(sl *SlowLog, val float64) { sl.CommitTime = val }),
	"Get_commit_ts_time":        floatParser(func(sl *SlowLog, val float64) { sl.GetCommitTsTime = val }),
	"Commit_backoff_time":       floatParser(func(sl *SlowLog, val float64) { sl.CommitBackoffTime = val }),
	"Resolve_lock_time":         floatParser(func(sl *SlowLog, val float64) { sl.ResolveLockTime = val }),
	"Local_latch_wait_time":     floatParser(func(sl *SlowLog, val float64) { sl.LocalLatchWaitTime = val }),
	"Write_keys":                intParser(func(sl *SlowLog, val int) { sl.WriteKeys = val }),
	"Write_size":                intParser(func(sl *SlowLog, val int) { sl.WriteSize = val }),
	"Prewrite_region":           intParser(func(sl *SlowLog, val int) { sl.PrewriteRegion = val }),
	"Txn_retry":                 intParser(func(sl *SlowLog, val int) { sl.TxnRetry = val }),
	"Cop_time":                  floatParser(func(sl *SlowLog, val float64) { sl.CopTime = val }),
	"Process_time":              floatParser(func(sl *SlowLog, val float64) { sl.ProcessTime = val }),
	"Wait_time":                 floatParser(func(sl *SlowLog, val float64) { sl.WaitTime = val }),
	"Backoff_time":              floatParser(func(sl *SlowLog, val float64) { sl.BackoffTime = val }),
	"LockKeys_time":             floatParser(func(sl *SlowLog, val float64) { sl.LockKeysTime = val }),
	"Request_count":             intParser(func(sl *SlowLog, val int) { sl.RequestCount = val }),
	"Total_keys":                intParser(func(sl *SlowLog, val int) { sl.TotalKeys = val }),
	"Process_keys":              intParser(func(sl *SlowLog, val int) { sl.ProcessKeys = val }),
	"Rocksdb_delete_skipped_count": intParser(func(sl *SlowLog, val int) {
		sl.RocksdbDeleteSkippedCount = val
	}),
	"Rocksdb_key_skipped_count": intParser(func(sl *SlowLog, val int) { sl.RocksdbKeySkippedCount = val }),
	"Rocksdb_block_cache_hit_count": intParser(func(sl *SlowLog, val int) {
		sl.RocksdbBlockCacheHitCount = val
	}),
	"Rocksdb_block_read_count": intParser(func(sl *SlowLog, val int) { sl.RocksdbBlockReadCount = val }),
	"Rocksdb_block_read_byte":  intParser(func(sl *SlowLog, val int) { sl.RocksdbBlockReadByte = val }),
	"Is_internal":              boolParser(func(sl *SlowLog, val bool) { sl.IsInternal = val }),
	"Cop_proc_avg":             floatParser(func(sl *SlowLog, val float64) { sl.CopProcAvg = val }),
	"Cop_proc_p90":             floatParser(func(sl *SlowLog, val float64) { sl.CopProcP90 = val }),
	"Cop_proc_max":             floatParser(func(sl *SlowLog, val float64) { sl.CopProcMax = val }),
	"Cop_proc_addr": func(sl *SlowLog, v string) error {
		sl.CopProcAddr = v
		return nil
	},
	"Cop_wait_avg": floatParser(func(sl *SlowLog, val float64) { sl.CopWaitAvg = val }),
	"Cop_wait_p90": floatParser(func(sl *SlowLog, val float64) { sl.CopWaitP90 = val }),
	"Cop_wait_max": floatParser(func(sl *SlowLog, val float64) { sl.CopWaitMax = val }),
	"Cop_wait_addr": func(sl *SlowLog, v string) error {
		sl.CopWaitAddr = v
		return nil
	},
	"Mem_max":                  intParser(func(sl *SlowLog, val int) { sl.MemMax = val }),
	"Disk_max":                 intParser(func(sl *SlowLog, val int) { sl.DiskMax = val }),
	"KV_total":                 floatParser(func(sl *SlowLog, val float64) { sl.KVTotal = val }),
	"PD_total":                 floatParser(func(sl *SlowLog, val float64) { sl.PDTotal = val }),
	"Backoff_total":            floatParser(func(sl *SlowLog, val float64) { sl.BackoffTotal = val }),
	"Write_sql_response_total": floatParser(func(sl *SlowLog, val float64) { sl.WriteSQLResponseTotal = val }),
	"Result_rows":              intParser(func(sl *SlowLog, val int) { sl.ResultRows = val }),
	"Prepared":                 boolParser(func(sl *SlowLog, val bool) { sl.Prepared = val }),
	"Succ":                     boolParser(func(sl *SlowLog, val bool) { sl.Succ = val }),
	"IsExplicitTxn":            boolParser(func(sl *SlowLog, val bool) { sl.IsExplicitTxn = val }),
	"IsWriteCacheTable":        boolParser(func(sl *SlowLog, val bool) { sl.IsWriteCacheTable = val }),
	"Plan_from_cache":          boolParser(func(sl *SlowLog, val bool) { sl.PlanFromCache = val }),
	"Plan_from_binding":        boolParser(func(sl *SlowLog, val bool) { sl.PlanFromBinding = val }),
	"Has_more_results":         boolParser(func(sl *SlowLog, val bool) { sl.HasMoreResults = val }),
	"Request_unit_read":        floatParser(func(sl *SlowLog, val float64) { sl.RequestUnitRead = val }),
	"Request_unit_write":       floatParser(func(sl *SlowLog, val float64) { sl.RequestUnitWrite = val }),
	"Time_queued_by_rc":        floatParser(func(sl *SlowLog, val float64) { sl.TimeQueuedByRC = val }),
	"Is_tiflash":               boolParser(func(sl *SlowLog, val bool) { sl.IsTiFlash = val }),
}

func intParser(setter func(sl *SlowLog, val int)) fieldParser {
	return func(sl *SlowLog, v string) error {
		val, err := strconv.Atoi(v)
		if err != nil {
			return fmt.Errorf("invalid value: %v", err)
		}
		setter(sl, val)
		return nil
	}
}

func floatParser(setter func(sl *SlowLog, val float64)) fieldParser {
	return func(sl *SlowLog, v string) error {
		val, err := strconv.ParseFloat(v, 64)
		if err != nil {
			return fmt.Errorf("invalid value: %v", err)
		}
		setter(sl, val)
		return nil
	}
}

func boolParser(setter func(sl *SlowLog, val bool)) fieldParser {
	return func(sl *SlowLog, v string) error {
		setter(sl, strToBool(v))
		return nil
	}
}

var handlers = []lineHandler{
	&timeLineHandler{},
	&userHostHandler{},
	&simpleKVHandler{prefix: "# Session_alias: ", fieldFunc: func(sl *SlowLog, v string) { sl.SessionAlias = v }},
	&simpleKVHandler{prefix: "# Backoff_types: ", fieldFunc: func(sl *SlowLog, v string) { sl.BackoffTypes = v }},
	&simpleKVHandler{prefix: "# DB: ", fieldFunc: func(sl *SlowLog, v string) { sl.DB = v }},
	&simpleKVHandler{prefix: "# Index_names: ", fieldFunc: func(sl *SlowLog, v string) { sl.IndexNames = v }},
	&simpleKVHandler{prefix: "# Digest: ", fieldFunc: func(sl *SlowLog, v string) { sl.Digest = v }},
	&simpleKVHandler{prefix: "# Stats: ", fieldFunc: func(sl *SlowLog, v string) { sl.Stats = v }},
	&simpleKVHandler{prefix: "# Warnings: ", fieldFunc: func(sl *SlowLog, v string) { sl.Warnings = v }},
	&simpleKVHandler{prefix: "# Cop_backoff_", fieldFunc: func(sl *SlowLog, v string) { sl.BackoffDetail = v }},
	&simpleKVHandler{prefix: "# Plan: ", fieldFunc: func(sl *SlowLog, v string) { sl.Plan = v }},
	&simpleKVHandler{prefix: "# Resource_group: ", fieldFunc: func(sl *SlowLog, v string) { sl.ResourceGroup = v }},
	&simpleKVHandler{prefix: "# Plan_digest: ", fieldFunc: func(sl *SlowLog, v string) { sl.PlanDigest = v }},
	&simpleKVHandler{prefix: "# Binary_plan: ", fieldFunc: func(sl *SlowLog, v string) { sl.BinaryPlan = v }},
	&simpleKVHandler{prefix: "# Prev_stmt: ", fieldFunc: func(sl *SlowLog, v string) { sl.PrevStmt = v }},
	&simpleKVHandler{prefix: "# Serverless_tenant_ID: ", fieldFunc: func(sl *SlowLog, v string) { sl.ServerlessTenantID = v }},
	&simpleKVHandler{prefix: "# Serverless_project_ID: ", fieldFunc: func(sl *SlowLog, v string) { sl.ServerlessProjectID = v }},
	&simpleKVHandler{prefix: "# Serverless_cluster_ID: ", fieldFunc: func(sl *SlowLog, v string) { sl.ServerlessClusterID = v }},
	&simpleKVHandler{prefix: "# Worker_role: ", fieldFunc: func(sl *SlowLog, v string) { sl.WorkerRole = v }},
	&complexKVHandler{},
	&queryHandler{},
	&ignoreHandler{},
}

func parseLine(sl *SlowLog, line string) error {
	if line == "" {
		return nil
	}

	for _, handler := range handlers {
		if handler.canHandle(line) {
			return handler.handle(sl, line)
		}
	}

	return fmt.Errorf("invalid line: %s", truncateForLog(line, maxLogSnippetLen))
}

// strToBool converts a string to boolean
func strToBool(v string) bool {
	return v == "1" || strings.ToLower(v) == "true"
}

var (
	vectorLogPrefixReg = regexp.MustCompile(`ip-[^;]+`)
	splitByColonReg    = regexp.MustCompile(`([^:]+):\s*([^\s]+)`)
)

// Close closes the underlying reader
func (p *StreamParser) Close() error {
	if closer, ok := p.src.(io.Closer); ok {
		return closer.Close()
	}
	return nil
}

// streamFileToRoller downloads one source file and appends parsed records to the shared rolling uploader.
func (t *Task) streamFileToRoller(ctx context.Context, key string, roller *rollingGzipUploader) (sourceFileStats, error) {
	var stats sourceFileStats
	defer func() {
		if r := recover(); r != nil {
			stack := debug.Stack()
			log.Fatal("PANIC OCCURRED",
				zap.Any("reason", r),
				zap.String("stack", string(stack)),
			)
		}
	}()

	s3Key := filepath.Base(key)
	if strings.HasSuffix(s3Key, ".gz") {
		s3Key = s3Key[:len(s3Key)-3]
	}

	localFilePath := filepath.Join("/tmp", t.tenantId, t.clusterId, s3Key)
	defer func() {
		if _, err := os.Stat(localFilePath); err == nil {
			if err := os.Remove(localFilePath); err != nil {
				log.Error("Failed to remove temporary file",
					zap.String("path", localFilePath),
					zap.Error(err))
			}
		}
	}()

	if err := s3sem.Acquire(ctx, 1); err != nil {
		log.Error("failed to acquire semaphore", zap.Error(err))
		return stats, err
	}
	defer s3sem.Release(1)

	downloadStarted := time.Now()
	if err := t.downloadOSSFile(key, localFilePath); err != nil {
		return stats, fmt.Errorf("fail to download: %w", err)
	}
	stats.DownloadDuration = time.Since(downloadStarted)
	if fileInfo, err := os.Stat(localFilePath); err == nil {
		stats.DownloadBytes = fileInfo.Size()
	}

	processStarted := time.Now()
	if t.logKind == logKindStatement {
		records, err := t.streamStatementFileToRoller(localFilePath, key, roller)
		stats.ProcessDuration = time.Since(processStarted)
		stats.Records = records
		return stats, err
	}
	records, err := t.streamSlowlogFileToRoller(localFilePath, key, roller)
	stats.ProcessDuration = time.Since(processStarted)
	stats.Records = records
	return stats, err
}

func (t *Task) streamSlowlogFileToRoller(localFilePath, key string, roller *rollingGzipUploader) (int, error) {
	file, err := os.Open(localFilePath)
	if err != nil {
		return 0, err
	}
	defer file.Close()

	parser := NewStreamParser(file)
	records := 0
	for {
		record, err := parser.ReadNext()
		if err != nil {
			log.Warn("slowlog parse failed",
				zap.String("file", key),
				zap.String("error", truncateForLog(err.Error(), maxLogSnippetLen)),
			)
			processingErrors.Inc()
			continue
		}
		if record == nil {
			break
		}
		if t.filterTenantID != "" && record.ServerlessTenantID != t.filterTenantID {
			continue
		}
		if !isValidEventTime(record.Time) {
			log.Warn("slowlog record missing # Time, skipping",
				zap.String("file", key),
				zap.String("query", truncateForLog(record.Query, maxLogSnippetLen)),
			)
			continue
		}
		if err := roller.Append(*record); err != nil {
			return records, fmt.Errorf("append to rolling uploader: %w", err)
		}
		recordsRead.Inc()
		records++
	}
	return records, nil
}

func (t *Task) streamStatementFileToRoller(localFilePath, key string, roller *rollingGzipUploader) (int, error) {
	file, err := os.Open(localFilePath)
	if err != nil {
		return 0, err
	}
	defer file.Close()

	parser := NewStatementStreamParser(file)
	records := 0
	for {
		record, err := parser.ReadNext()
		if err != nil {
			log.Warn("statement parse failed",
				zap.String("file", key),
				zap.String("error", truncateForLog(err.Error(), maxLogSnippetLen)),
			)
			processingErrors.Inc()
			continue
		}
		if record == nil {
			break
		}
		if t.filterTenantID != "" && record.ServerlessTenantID != t.filterTenantID {
			continue
		}
		eventTime := record.eventTime()
		if !isValidEventTime(eventTime) {
			log.Warn("statement record missing begin/end time, skipping",
				zap.String("file", key),
			)
			continue
		}
		if err := roller.AppendStatement(record.Raw, eventTime); err != nil {
			return records, fmt.Errorf("append statement to rolling uploader: %w", err)
		}
		recordsRead.Inc()
		records++
	}
	return records, nil
}

// CollectSlowLogs collects all slow logs from a file without uploading
func (t *Task) CollectSlowLogs(ctx context.Context, key string) ([]SlowLog, error) {
	defer func() {
		if r := recover(); r != nil {
			stack := debug.Stack()
			log.Fatal("PANIC OCCURRED",
				zap.Any("reason", r),
				zap.String("stack", string(stack)),
			)
		}
	}()

	s3Key := filepath.Base(key)
	if strings.HasSuffix(s3Key, ".gz") {
		s3Key = s3Key[:len(s3Key)-3]
	}

	localFilePath := filepath.Join("/tmp", t.tenantId, t.clusterId, s3Key)
	defer func() {
		if _, err := os.Stat(localFilePath); err == nil {
			if err := os.Remove(localFilePath); err != nil {
				log.Error("Failed to remove temporary file",
					zap.String("path", localFilePath),
					zap.Error(err))
			}
		}
	}()

	if err := s3sem.Acquire(ctx, 1); err != nil {
		log.Error("failed to acquire semaphore", zap.Error(err))
		return nil, err
	}
	defer s3sem.Release(1)

	if err := t.downloadOSSFile(key, localFilePath); err != nil {
		return nil, fmt.Errorf("fail to download: %w", err)
	}

	return t.collectSlowLogsFromFile(localFilePath)
}

func (t *Task) collectSlowLogsFromFile(localFilePath string) ([]SlowLog, error) {
	file, err := os.Open(localFilePath)
	if err != nil {
		return nil, err
	}
	defer file.Close()

	parser := NewStreamParser(file)
	var allSlowLogs []SlowLog

	for {
		record, err := parser.ReadNext()
		if err != nil {
			log.Warn("parse failed", zap.Error(err))
			processingErrors.Inc()
			continue
		}
		if record == nil {
			break
		}
		if !isValidEventTime(record.Time) {
			log.Warn("slowlog record missing # Time, skipping",
				zap.String("query", truncateForLog(record.Query, maxLogSnippetLen)),
			)
			continue
		}

		allSlowLogs = append(allSlowLogs, *record)
		recordsRead.Inc()
	}

	return allSlowLogs, nil
}

func (t *Task) ThreadTask(ctx context.Context, key string, fileLock *sync.Mutex) (int, error) {
	defer func() {
		if r := recover(); r != nil {
			stack := debug.Stack()
			log.Fatal("PANIC OCCURRED",
				zap.Any("reason", r),
				zap.String("stack", string(stack)),
			)
		} else {
			log.Info("Task completed")
		}
	}()

	s3Key := filepath.Base(key)
	if strings.HasSuffix(s3Key, ".gz") {
		s3Key = s3Key[:len(s3Key)-3]
	}

	localFilePath := filepath.Join("/tmp", t.tenantId, t.clusterId, s3Key)
	defer func() {
		if _, err := os.Stat(localFilePath); err == nil {
			if err := os.Remove(localFilePath); err != nil {
				log.Error("Failed to remove temporary file",
					zap.String("path", localFilePath),
					zap.Error(err))
			}
		}
	}()

	if err := s3sem.Acquire(ctx, 1); err != nil {
		log.Error("failed to acquire semaphore", zap.Error(err))
		return -1, err
	}
	defer s3sem.Release(1)

	if err := t.downloadOSSFile(key, localFilePath); err != nil {
		return 0, fmt.Errorf("fail to download: %w", err)
	}

	return t.processLogFile(localFilePath, fileLock)
}

func (t *Task) downloadOSSFile(key, path string) (retErr error) {
	ctx, cancel := context.WithTimeout(t.ctx, 30*time.Second)
	defer cancel()

	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
		return fmt.Errorf("failed to create directory: %w", err)
	}

	var result *storage.Object
	var err error
	for i := 0; i < 3; i++ {
		result, err = t.sourceStorage.Object(ctx, t.sourceBucket, key)
		if err == nil {
			break
		}
		time.Sleep(time.Second * time.Duration(i+1))
	}
	if err != nil {
		return fmt.Errorf("after 3 attempts, failed to get object: %w", err)
	}

	defer func() {
		if closeErr := result.Reader.Close(); closeErr != nil && retErr == nil {
			retErr = fmt.Errorf("failed to close reader: %w", closeErr)
		}
	}()

	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
		return fmt.Errorf("failed to create directory: %w", err)
	}

	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	if err != nil {
		return fmt.Errorf("failed to create file: %w", err)
	}
	defer func() {
		if closeErr := f.Close(); closeErr != nil && retErr == nil {
			retErr = fmt.Errorf("failed to close file: %w", closeErr)
		}
	}()

	if _, err := io.Copy(f, result.Reader); err != nil {
		return fmt.Errorf("failed to write file: %w", err)
	}

	return nil
}

func (t *Task) processLogFile(localFilePath string, lock *sync.Mutex) (int, error) {
	file, err := os.Open(localFilePath)
	if err != nil {
		return 0, err
	}
	defer file.Close()

	parser := NewStreamParser(file)
	var (
		logCount     int
		batch        []SlowLog
		lastErr      error
		writtenCount int
	)

	var fileId = 0
	now := time.Now().Local()
	fileNamePrefix := now.Format("200601021504")
	const batchSize = 1000

	for {
		record, err := parser.ReadNext()
		if err != nil {
			fmt.Printf("pparse failed: %v", err)
			processingErrors.Inc()
			lastErr = err
			continue
		}
		if record == nil {
			break
		}
		if !isValidEventTime(record.Time) {
			log.Warn("slowlog record missing # Time, skipping",
				zap.String("query", truncateForLog(record.Query, maxLogSnippetLen)),
			)
			continue
		}

		batch = append(batch, *record)
		logCount++
		recordsRead.Inc()

		if len(batch) >= batchSize {
			fileId++
			tmpDir := os.TempDir()
			tmpFile := filepath.Join(tmpDir, fmt.Sprintf("slowlog_%s_%d_%d_%d", fileNamePrefix, t.id, fileId, time.Now().UnixNano()))
			if err := t.processBatch(batch, tmpFile, lock); err != nil {
				if lastErr == nil {
					lastErr = fmt.Errorf("batch process failed: %w", err)
				}
				processingErrors.Inc()
				continue
			}
			writtenCount += len(batch)
			recordsWritten.Add(float64(len(batch)))
			// compressFile returns tmpFile.gz, which is already removed by processBatch's defer
			// So we don't need to track it for cleanup
			batch = nil
		}
	}
	if len(batch) > 0 {
		fileId++
		tmpDir := os.TempDir()
		tmpFile := filepath.Join(tmpDir, fmt.Sprintf("slowlog_%s_%d_%d_%d", fileNamePrefix, t.id, fileId, time.Now().UnixNano()))
		if err := t.processBatch(batch, tmpFile, lock); err != nil && lastErr == nil {
			lastErr = fmt.Errorf("process last failed: %w", err)
			processingErrors.Inc()
		} else {
			writtenCount += len(batch)
			recordsWritten.Add(float64(len(batch)))
			// compressFile returns tmpFile.gz, which is already removed by processBatch's defer
			// So we don't need to track it for cleanup
		}
	}

	if logCount != writtenCount {
		mismatch := logCount - writtenCount
		recordsMismatch.Add(float64(mismatch))
		fmt.Printf("warning: read %d, write %d,unmatch %d",
			logCount, writtenCount, mismatch)
	}

	// Note: compressedFiles cleanup is no longer needed because:
	// 1. processBatch already removes the compressed file with defer os.Remove(gzFile)
	// 2. The original tmpFile is removed by compressFile function
	// So we can remove the cleanup logic here

	return logCount, lastErr
}

// cleanupTempFiles is no longer needed because processBatch already handles cleanup
// with defer os.Remove(gzFile)

func isValidEventTime(t time.Time) bool {
	return !t.IsZero()
}

// batchTimeRange returns start and end time from a batch of slow logs (UTC, format 20060102150405).
// Records with zero Time (unparsed # Time line) are ignored.
func batchTimeRange(batch []SlowLog) (startTime, endTime time.Time) {
	for _, log := range batch {
		if !isValidEventTime(log.Time) {
			continue
		}
		if startTime.IsZero() || log.Time.Before(startTime) {
			startTime = log.Time
		}
		if endTime.IsZero() || log.Time.After(endTime) {
			endTime = log.Time
		}
	}
	return startTime, endTime
}

func (t *Task) processBatch(batch []SlowLog, tmpFile string, lock *sync.Mutex) error {
	if err := t.writeBatch(tmpFile, batch, lock); err != nil {
		return err
	}

	gzPath, err := t.compressFile(tmpFile)
	if err != nil {
		return fmt.Errorf("fail to compress: %w", err)
	}
	defer func() {
		if err := os.Remove(gzPath); err != nil && !os.IsNotExist(err) {
			log.Warn("failed to remove gzip tmp file", zap.String("path", gzPath), zap.Error(err))
		}
	}()

	startTime, endTime := batchTimeRange(batch)
	startTimeStr := startTime.UTC().Format("20060102150405")
	endTimeStr := endTime.UTC().Format("20060102150405")
	md5HashGz, err := calculateFileMD5(gzPath)
	if err != nil {
		return fmt.Errorf("fail to calculate MD5 of gzip file: %w", err)
	}
	md5PrefixGz := md5HashGz[:8]
	filename := fmt.Sprintf("slowlog_%s_%s_%s.gz", startTimeStr, endTimeStr, md5PrefixGz)

	data, err := os.ReadFile(gzPath)
	if err != nil {
		return fmt.Errorf("fail to read gzip file %s: %w", gzPath, err)
	}
	if len(data) == 0 {
		return fmt.Errorf("gzip file is empty: %s", gzPath)
	}

	fullPath := filepath.Join(t.targetFilePath, filename)
	if err := t.targetStorage.SaveObject(t.ctx, t.targetBucket, fullPath, bytes.NewReader(data)); err != nil {
		return fmt.Errorf("fail to upload: %w", err)
	}
	return nil
}

func (t *Task) compressFile(src string) (string, error) {
	dest := src + ".gz"

	srcFile, err := os.Open(src)
	if err != nil {
		return "", err
	}
	defer srcFile.Close()

	gzFile, err := os.Create(dest)
	if err != nil {
		return "", err
	}

	gzWriter := gzip.NewWriter(gzFile)
	if _, err := io.Copy(gzWriter, srcFile); err != nil {
		gzWriter.Close()
		gzFile.Close()
		os.Remove(dest)
		return "", err
	}
	if err := gzWriter.Close(); err != nil {
		gzFile.Close()
		os.Remove(dest)
		return "", fmt.Errorf("fail to close gzip writer: %w", err)
	}
	if err := gzFile.Sync(); err != nil {
		gzFile.Close()
		os.Remove(dest)
		return "", fmt.Errorf("fail to sync gzip file: %w", err)
	}
	if err := gzFile.Close(); err != nil {
		os.Remove(dest)
		return "", fmt.Errorf("fail to close gzip file: %w", err)
	}

	if err := os.Remove(src); err != nil && !os.IsNotExist(err) {
		log.Warn("failed to remove tmp file after compress", zap.String("path", src), zap.Error(err))
	}

	return dest, nil
}

func (t *Task) writeBatch(tempFile string, batch []SlowLog, lock *sync.Mutex) error {
	lock.Lock()
	defer lock.Unlock()

	file, err := os.OpenFile(tempFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	defer file.Close()

	//enc := json.NewEncoder(file)
	//return enc.Encode(batch)
	writer := csv.NewWriter(file)
	defer writer.Flush()
	if stat, err := file.Stat(); err == nil && stat.Size() == 0 {
		if err := writer.Write(slowLogCSVHeader()); err != nil {
			return err
		}
	}

	for _, log := range batch {
		if err := appendSlowLogCSVRow(writer, log); err != nil {
			return err
		}
	}
	return nil
}

type Task struct {
	ctx            context.Context
	sourceBucket   string
	targetBucket   string
	tenantId       string
	clusterId      string
	sourceStorage  storage.ListStorage
	targetStorage  storage.ListStorage
	targetFilePath string
	id             int
	filterTenantID string
	logKind        string
}

// ProcessAndUploadSlowLogs processes and uploads slow logs from a file (kept for single-file use).
func (t *Task) ProcessAndUploadSlowLogs(ctx context.Context, key string) error {
	logs, err := t.CollectSlowLogs(ctx, key)
	if err != nil {
		return err
	}
	return t.uploadCollectedSlowLogs(logs)
}

// uploadCollectedSlowLogs uploads collected slow logs in batches of batchSize (500) to avoid many tiny files.
func (t *Task) uploadCollectedSlowLogs(logs []SlowLog) error {
	const batchSize = 500
	for i := 0; i < len(logs); i += batchSize {
		end := i + batchSize
		if end > len(logs) {
			end = len(logs)
		}
		batch := logs[i:end]
		if err := t.uploadOneBatch(batch); err != nil {
			log.Error("batch process failed", zap.Error(err))
			processingErrors.Inc()
			continue
		}
		recordsWritten.Add(float64(len(batch)))
		runtime.GC()
	}
	return nil
}

// uploadOneBatch uploads a single batch (used when accumulating across files).
func (t *Task) uploadOneBatch(batch []SlowLog) error {
	if len(batch) == 0 {
		return nil
	}
	tmpDir := os.TempDir()
	tmpFile := filepath.Join(tmpDir, fmt.Sprintf("slowlog_%d_%d", t.id, time.Now().UnixNano()))
	return t.processBatch(batch, tmpFile, &sync.Mutex{})
}

// uploadUnifiedFile uploads all slow logs as a single file with new naming format
func uploadUnifiedFile(ctx context.Context, targetStorage storage.ListStorage, targetBucket string, targetFilePath string, slowLogs []SlowLog) error {
	if len(slowLogs) == 0 {
		return fmt.Errorf("no slow logs to upload")
	}

	validLogs := make([]SlowLog, 0, len(slowLogs))
	for _, log := range slowLogs {
		if isValidEventTime(log.Time) {
			validLogs = append(validLogs, log)
		}
	}
	if len(validLogs) == 0 {
		return fmt.Errorf("no slow logs with valid time to upload")
	}

	startTime, endTime := batchTimeRange(validLogs)
	if !isValidEventTime(startTime) || !isValidEventTime(endTime) {
		return fmt.Errorf("failed to derive time range from slow logs")
	}

	// Format times as UTC, precise to second: 20260203074701
	startTimeStr := startTime.UTC().Format("20060102150405")
	endTimeStr := endTime.UTC().Format("20060102150405")

	// Create temporary CSV file
	tmpDir := os.TempDir()
	tmpFile := filepath.Join(tmpDir, fmt.Sprintf("slowlog_%s_%s", startTimeStr, endTimeStr))
	defer os.Remove(tmpFile)

	// Write all slow logs to CSV
	if err := writeSlowLogsToCSV(tmpFile, validLogs); err != nil {
		return fmt.Errorf("failed to write CSV: %w", err)
	}

	// Compress file
	gzFile, err := compressFile(tmpFile)
	if err != nil {
		return fmt.Errorf("failed to compress: %w", err)
	}
	defer os.Remove(gzFile)

	// Calculate MD5 of compressed file (for filename)
	md5HashGz, err := calculateFileMD5(gzFile)
	if err != nil {
		return fmt.Errorf("failed to calculate MD5 of compressed file: %w", err)
	}
	md5PrefixGz := md5HashGz[:8]

	// Generate filename: slowlog_{Start_Time}_{End_Time}_{MD5_8}.gz
	filename := fmt.Sprintf("slowlog_%s_%s_%s.gz", startTimeStr, endTimeStr, md5PrefixGz)

	// Upload to target storage
	f, err := os.Open(gzFile)
	if err != nil {
		return fmt.Errorf("failed to open compressed file: %w", err)
	}
	defer f.Close()

	fullPath := filepath.Join(targetFilePath, filename)
	if err := targetStorage.SaveObject(ctx, targetBucket, fullPath, f); err != nil {
		return fmt.Errorf("failed to upload: %w", err)
	}

	log.Info("successfully uploaded unified file",
		zap.String("filename", filename),
		zap.Int("recordCount", len(validLogs)),
		zap.String("startTime", startTimeStr),
		zap.String("endTime", endTimeStr))

	recordsWritten.Add(float64(len(validLogs)))
	return nil
}

func writeSlowLogsToCSV(filename string, slowLogs []SlowLog) error {
	file, err := os.Create(filename)
	if err != nil {
		return err
	}
	defer file.Close()

	writer := csv.NewWriter(file)
	defer writer.Flush()

	// Write header
	header := []string{
		"Time", "Instance", "User", "Host", "ConnID", "SessionAlias",
		"ExecRetryCount", "ExecRetryTime", "QueryTime", "ParseTime",
		"CompileTime", "RewriteTime", "PreprocSubqueries", "PreprocSubqueriesTime",
		"OptimizeTime", "WaitTS", "PrewriteTime", "WaitPrewriteBinlogTime",
		"CommitTime", "GetCommitTsTime", "CommitBackoffTime", "BackoffTypes",
		"ResolveLockTime", "LocalLatchWaitTime", "WriteKeys", "WriteSize",
		"PrewriteRegion", "TxnRetry", "CopTime", "ProcessTime", "WaitTime",
		"BackoffTime", "LockKeysTime", "RequestCount", "TotalKeys", "ProcessKeys",
		"RocksdbDeleteSkippedCount", "RocksdbKeySkippedCount", "RocksdbBlockCacheHitCount",
		"RocksdbBlockReadCount", "RocksdbBlockReadByte", "DB", "IndexNames",
		"IsInternal", "Digest", "Stats", "CopProcAvg", "CopProcP90", "CopProcMax",
		"CopProcAddr", "CopWaitAvg", "CopWaitP90", "CopWaitMax", "CopWaitAddr",
		"MemMax", "DiskMax", "KVTotal", "PDTotal", "BackoffTotal",
		"WriteSQLResponseTotal", "ResultRows", "Warnings", "BackoffDetail",
		"Prepared", "Succ", "IsExplicitTxn", "IsWriteCacheTable", "PlanFromCache",
		"PlanFromBinding", "HasMoreResults", "ResourceGroup", "RequestUnitRead",
		"RequestUnitWrite", "TimeQueuedByRC", "Plan", "PlanDigest", "BinaryPlan",
		"PrevStmt", "Query", "ServerlessTenantID", "ServerlessProjectID",
		"ServerlessClusterID", "WorkerRole", "IsTiFlash", "TxnStartTs",
	}
	if err := writer.Write(header); err != nil {
		return err
	}

	// Write records
	for _, log := range slowLogs {
		record := slowLogToCSVRecord(log)
		if err := writer.Write(record); err != nil {
			return err
		}
	}

	return nil
}

func slowLogToCSVRecord(log SlowLog) []string {
	return []string{
		log.Time.Format(time.RFC3339),
		log.Instance,
		log.User,
		log.Host,
		strconv.Itoa(log.ConnID),
		log.SessionAlias,
		strconv.Itoa(log.ExecRetryCount),
		strconv.FormatFloat(log.ExecRetryTime, 'f', -1, 64),
		strconv.FormatFloat(log.QueryTime, 'f', -1, 64),
		strconv.FormatFloat(log.ParseTime, 'f', -1, 64),
		strconv.FormatFloat(log.CompileTime, 'f', -1, 64),
		strconv.FormatFloat(log.RewriteTime, 'f', -1, 64),
		strconv.Itoa(log.PreprocSubqueries),
		strconv.FormatFloat(log.PreprocSubqueriesTime, 'f', -1, 64),
		strconv.FormatFloat(log.OptimizeTime, 'f', -1, 64),
		strconv.FormatFloat(log.WaitTS, 'f', -1, 64),
		strconv.FormatFloat(log.PrewriteTime, 'f', -1, 64),
		strconv.FormatFloat(log.WaitPrewriteBinlogTime, 'f', -1, 64),
		strconv.FormatFloat(log.CommitTime, 'f', -1, 64),
		strconv.FormatFloat(log.GetCommitTsTime, 'f', -1, 64),
		strconv.FormatFloat(log.CommitBackoffTime, 'f', -1, 64),
		log.BackoffTypes,
		strconv.FormatFloat(log.ResolveLockTime, 'f', -1, 64),
		strconv.FormatFloat(log.LocalLatchWaitTime, 'f', -1, 64),
		strconv.Itoa(log.WriteKeys),
		strconv.Itoa(log.WriteSize),
		strconv.Itoa(log.PrewriteRegion),
		strconv.Itoa(log.TxnRetry),
		strconv.FormatFloat(log.CopTime, 'f', -1, 64),
		strconv.FormatFloat(log.ProcessTime, 'f', -1, 64),
		strconv.FormatFloat(log.WaitTime, 'f', -1, 64),
		strconv.FormatFloat(log.BackoffTime, 'f', -1, 64),
		strconv.FormatFloat(log.LockKeysTime, 'f', -1, 64),
		strconv.Itoa(log.RequestCount),
		strconv.Itoa(log.TotalKeys),
		strconv.Itoa(log.ProcessKeys),
		strconv.Itoa(log.RocksdbDeleteSkippedCount),
		strconv.Itoa(log.RocksdbKeySkippedCount),
		strconv.Itoa(log.RocksdbBlockCacheHitCount),
		strconv.Itoa(log.RocksdbBlockReadCount),
		strconv.Itoa(log.RocksdbBlockReadByte),
		log.DB,
		log.IndexNames,
		strconv.FormatBool(log.IsInternal),
		log.Digest,
		log.Stats,
		strconv.FormatFloat(log.CopProcAvg, 'f', -1, 64),
		strconv.FormatFloat(log.CopProcP90, 'f', -1, 64),
		strconv.FormatFloat(log.CopProcMax, 'f', -1, 64),
		log.CopProcAddr,
		strconv.FormatFloat(log.CopWaitAvg, 'f', -1, 64),
		strconv.FormatFloat(log.CopWaitP90, 'f', -1, 64),
		strconv.FormatFloat(log.CopWaitMax, 'f', -1, 64),
		log.CopWaitAddr,
		strconv.Itoa(log.MemMax),
		strconv.Itoa(log.DiskMax),
		strconv.FormatFloat(log.KVTotal, 'f', -1, 64),
		strconv.FormatFloat(log.PDTotal, 'f', -1, 64),
		strconv.FormatFloat(log.BackoffTotal, 'f', -1, 64),
		strconv.FormatFloat(log.WriteSQLResponseTotal, 'f', -1, 64),
		strconv.Itoa(log.ResultRows),
		log.Warnings,
		log.BackoffDetail,
		strconv.FormatBool(log.Prepared),
		strconv.FormatBool(log.Succ),
		strconv.FormatBool(log.IsExplicitTxn),
		strconv.FormatBool(log.IsWriteCacheTable),
		strconv.FormatBool(log.PlanFromCache),
		strconv.FormatBool(log.PlanFromBinding),
		strconv.FormatBool(log.HasMoreResults),
		log.ResourceGroup,
		strconv.FormatFloat(log.RequestUnitRead, 'f', -1, 64),
		strconv.FormatFloat(log.RequestUnitWrite, 'f', -1, 64),
		strconv.FormatFloat(log.TimeQueuedByRC, 'f', -1, 64),
		log.Plan,
		log.PlanDigest,
		log.BinaryPlan,
		log.PrevStmt,
		log.Query,
		log.ServerlessTenantID,
		log.ServerlessProjectID,
		log.ServerlessClusterID,
		log.WorkerRole,
		strconv.FormatBool(log.IsTiFlash),
		strconv.FormatUint(log.TxnStartTs, 10),
	}
}

func calculateFileMD5(filename string) (string, error) {
	file, err := os.Open(filename)
	if err != nil {
		return "", err
	}
	defer file.Close()

	hash := md5.New()
	if _, err := io.Copy(hash, file); err != nil {
		return "", err
	}

	return hex.EncodeToString(hash.Sum(nil)), nil
}

func compressFile(src string) (string, error) {
	dest := src + ".gz"

	srcFile, err := os.Open(src)
	if err != nil {
		return "", err
	}
	defer srcFile.Close()

	gzFile, err := os.Create(dest)
	if err != nil {
		return "", err
	}

	gzWriter := gzip.NewWriter(gzFile)
	if _, err := io.Copy(gzWriter, srcFile); err != nil {
		gzWriter.Close()
		gzFile.Close()
		os.Remove(dest)
		return "", err
	}
	if err := gzWriter.Close(); err != nil {
		gzFile.Close()
		os.Remove(dest)
		return "", fmt.Errorf("fail to close gzip writer: %w", err)
	}
	if err := gzFile.Sync(); err != nil {
		gzFile.Close()
		os.Remove(dest)
		return "", fmt.Errorf("fail to sync gzip file: %w", err)
	}
	if err := gzFile.Close(); err != nil {
		os.Remove(dest)
		return "", fmt.Errorf("fail to close gzip file: %w", err)
	}
	return dest, nil
}