package server

import (
	"context"
	stderrors "errors"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	"golang.org/x/sys/unix"
	"gopkg.in/natefinch/lumberjack.v2"
)

// Default values for push trigger configuration.
// These are vars (not const) to allow tests to modify them.
var (
	defaultCheckLogDuration = 5 * time.Second
	defaultCleanLogDuration = 10 * time.Second

	defaultLogHandler LogHandler = &NoPushHandler{}
)

const (
	defaultFilesPerPush = 3
	// defaultPushConcurrency is the default number of concurrent workers for pushing logs.
	defaultPushConcurrency = 4

	defaultAuditLogFileName = "tidb-audit.log"
	defaultSlowLogFileName  = "slowlog"

	// for `date-partitioned` path,
	// a `{{.date}}` placeholder will be replaced with the current date for the log file.
	dateTmplPlaceholder = "{{.date}}"

	// defaultMaxSize is the default file size threshold (in MiB) for triggering rotation and push in agent mode.
	defaultMaxSize = 10
	// defaultMaxLifetime is the default time interval (in seconds) for triggering rotation in agent mode.
	defaultMaxLifetime = 86400
)

type Server struct {
	vp     *viper.Viper
	logger *zap.Logger

	Config

	handler LogHandler

	skipCheckConn bool

	// state
	pushCountInShutdown int

	// Agent mode: tracks when last rotation occurred for maxLifetime calculation
	lastRotateTime time.Time

	// nowFunc is used for testing to mock time
	nowFunc func() time.Time
}

// getCheckInterval returns the configured check interval or default.
func (a *Server) getCheckInterval() time.Duration {
	if a.PushTrigger != nil && a.PushTrigger.CheckInterval > 0 {
		return a.PushTrigger.CheckInterval
	}
	return defaultCheckLogDuration
}

// getCleanInterval returns the configured clean interval or default.
func (a *Server) getCleanInterval() time.Duration {
	if a.PushTrigger != nil && a.PushTrigger.CleanInterval > 0 {
		return a.PushTrigger.CleanInterval
	}
	return defaultCleanLogDuration
}

// getFilesPerPush returns the configured files per push or default.
func (a *Server) getFilesPerPush() int {
	if a.PushTrigger != nil && a.PushTrigger.FilesPerPush != nil && *a.PushTrigger.FilesPerPush > 0 {
		return *a.PushTrigger.FilesPerPush
	}
	return defaultFilesPerPush
}

// getNow returns the current time, using nowFunc if set (for testing).
func (a *Server) getNow() time.Time {
	if a.nowFunc != nil {
		return a.nowFunc()
	}
	return time.Now()
}

// getMaxSize returns the configured max size (in MiB) or default.
// Returns 0 if explicitly disabled.
func (a *Server) getMaxSize() int {
	if a.PushTrigger == nil || a.PushTrigger.MaxSize == nil {
		return defaultMaxSize
	}
	return *a.PushTrigger.MaxSize // 0 = disabled
}

// getMaxLifetime returns the configured max lifetime (in seconds).
// Returns 0 if explicitly disabled.
func (a *Server) getMaxLifetime() int {
	if a.PushTrigger == nil || a.PushTrigger.MaxLifetime == nil {
		return defaultMaxLifetime
	}
	return *a.PushTrigger.MaxLifetime // 0 = disabled
}

// shouldRotateInAgentMode determines if rotation should occur in agent mode.
// Returns (shouldRotate, reason).
func (a *Server) shouldRotateInAgentMode() (bool, string) {
	if Mode(a.Mode) != ModeAgent {
		return false, ""
	}

	maxSize := a.getMaxSize()
	maxLifetime := a.getMaxLifetime()

	fname := filepath.Join(a.LogDir, a.AuditLogFileName)
	stat, err := os.Stat(fname)

	// File doesn't exist - update lastRotateTime to prevent repeated stat calls
	if err != nil {
		if a.lastRotateTime.IsZero() {
			a.lastRotateTime = a.getNow()
		}
		return false, ""
	}

	// Size-based rotation (maxSize > 0 means enabled)
	if maxSize > 0 && stat.Size() >= int64(maxSize)*1024*1024 {
		return true, fmt.Sprintf("size %d MiB >= %d MiB", stat.Size()/(1024*1024), maxSize)
	}

	// Lifetime-based rotation (maxLifetime > 0 means enabled)
	if maxLifetime > 0 {
		if a.lastRotateTime.IsZero() {
			a.lastRotateTime = a.getNow()
			return false, ""
		}
		elapsed := a.getNow().Sub(a.lastRotateTime)
		if elapsed >= time.Duration(maxLifetime)*time.Second {
			// Empty file: update lastRotateTime, skip rotation
			if stat.Size() == 0 {
				a.lastRotateTime = a.getNow()
				return false, ""
			}
			return true, fmt.Sprintf("elapsed %v >= %d s", elapsed.Round(time.Second), maxLifetime)
		}
	}

	return false, ""
}

// New create a Server.
// if the Viper use config file, Server will watch config and reload config.
func New(vp *viper.Viper) (*Server, error) {
	vp.SetDefault("auditLogFileName", defaultAuditLogFileName)
	vp.SetDefault("slowLogFileName", defaultSlowLogFileName)
	vp.SetDefault("pushConcurrency", defaultPushConcurrency)
	vp.SetDefault("slowLogMaxBackups", 1)
	vp.SetDefault("cleanOldestLogPercent", 0.8)
	// we need to set the default value for MetricsPort
	// to avoid no metrics for Pods taken from Standby TidbGroup.
	vp.SetDefault("metricsPort", 9090)
	// default 4000 for TiDB
	vp.SetDefault("producerPort", 4000)
	// default to rename strategy for backward compatibility
	vp.SetDefault("rotationStrategy", string(RotationStrategyRename))
	// default to tidb mode for backward compatibility
	vp.SetDefault("mode", string(ModeTiDB))

	err := vp.ReadInConfig()
	if err != nil {
		return nil, errors.Annotate(err, "failed to read config")
	}
	vp.WatchConfig()

	a := &Server{
		vp:     vp,
		logger: log.L(),
	}
	a.loadConfig()

	return a, nil
}

func (a *Server) serveMetrics(port int) error {
	registry := prometheus.NewRegistry()
	register(registry)

	handler := promhttp.InstrumentMetricHandler(registry, promhttp.HandlerFor(registry, promhttp.HandlerOpts{}))
	http.Handle("/metrics", handler)
	//nolint:gosec //only serve internal metrics
	return http.ListenAndServe(":"+strconv.Itoa(port), nil)
}

// loadConfig load config from viper and rebuild the log handler.
func (a *Server) loadConfig() {
	tmp := Config{}
	err := a.vp.Unmarshal(&tmp)
	if err != nil {
		errorCounter.WithLabelValues("unmarshal_config").Inc()
		a.logger.Error("failed to unmarshal config", zap.Error(err))
		return
	}

	// Normalize empty Mode to ModeTiDB for backward compatibility.
	// This handles the case where config explicitly sets mode: "" which would
	// override viper's default value.
	if tmp.Mode == "" {
		tmp.Mode = string(ModeTiDB)
	}

	// Normalize empty RotationStrategy to RotationStrategyRename for backward compatibility.
	if tmp.RotationStrategy == "" {
		tmp.RotationStrategy = string(RotationStrategyRename)
	}

	// Validate configuration values
	if err := validateMode(tmp.Mode); err != nil {
		errorCounter.WithLabelValues("invalid_config").Inc()
		a.logger.Error("invalid config", zap.Error(err))
		return
	}
	if err := validateRotationStrategy(tmp.RotationStrategy); err != nil {
		errorCounter.WithLabelValues("invalid_config").Inc()
		a.logger.Error("invalid config", zap.Error(err))
		return
	}

	// Validate push trigger config
	if err := validatePushTrigger(tmp.PushTrigger); err != nil {
		errorCounter.WithLabelValues("invalid_config").Inc()
		a.logger.Error("invalid config", zap.Error(err))
		return
	}
	// Validate subPath config
	if err := validateSubPath(tmp.SubPath, tmp.Mode); err != nil {
		errorCounter.WithLabelValues("invalid_config").Inc()
		a.logger.Error("invalid config", zap.Error(err))
		return
	}

	a.logger.Info("reload config", zap.Reflect("config", tmp.redactedForLog()))

	// Set Config first so that appendSuffix can access Application config
	a.Config = tmp

	if withSuffix, err := a.appendSuffix(a.BucketURL, os.Getenv("POD_NAME")); err == nil {
		a.logger.Info("append pod suffix", zap.String("url", withSuffix), zap.String("origin", a.BucketURL))
		a.BucketURL = withSuffix
	}

	// Do not replace if we are using a mock handler in test.
	if _, ok := a.handler.(*MockLogHandler); !ok {
		if a.handler != nil {
			a.handler.Close()
		}
		a.handler = a.buildLogHandler()
	}
}

func (a *Server) Run(ctx context.Context) error {
	if a.MetricsPort != 0 {
		go func() {
			err := a.serveMetrics(a.MetricsPort)
			if err != nil {
				log.Error("failed to serve metrics")
			}
		}()
	}

	configChangeCh := make(chan struct{}, 1)

	a.vp.OnConfigChange(func(e fsnotify.Event) {
		configChangeCh <- struct{}{}
	})

	handlerCh := make(chan struct{}, 1)
	checkLogTick := time.NewTicker(a.getCheckInterval())
	cleanLogTick := time.NewTicker(a.getCleanInterval())
	defer checkLogTick.Stop()
	defer cleanLogTick.Stop()

	for {
		select {
		case <-configChangeCh:
			//nolint:contextcheck //do not cancel from this outer context
			a.loadConfig()
			// Reset tickers with new intervals after config change
			checkLogTick.Reset(a.getCheckInterval())
			cleanLogTick.Reset(a.getCleanInterval())
		case <-handlerCh:
			if Mode(a.Mode) == ModeTiDB {
				// Only check for format changes in TiDB mode (TiDB-specific feature)
				a.rotateIfLogFormatChanged()
			} else if Mode(a.Mode) == ModeAgent {
				// Agent mode: rotate when maxSize or maxLifetime condition is met.
				if shouldRotate, reason := a.shouldRotateInAgentMode(); shouldRotate {
					a.logger.Debug("rotating in agent mode", zap.String("reason", reason))
					_, err := a.rotateIfNeed(a.LogDir, a.AuditLogFileName)
					if err != nil {
						a.logger.Error("failed to rotate log file", zap.Error(err))
					}
					// Always update lastRotateTime when shouldRotate=true
					// Prevents repeated attempts if rotateIfNeed returns false
					a.lastRotateTime = a.getNow()
				}
			}

			more, _, err := a.pushLog(ctx, a.handler, a.getFilesPerPush())
			if err != nil {
				a.logger.Error("failed to push log", zap.Error(err))
				continue
			}

			if more {
				handlerCh <- struct{}{}
			}
		case <-checkLogTick.C:
			select {
			case handlerCh <- struct{}{}:
			default:
			}
		case <-cleanLogTick.C:
			err := a.cleanSlowlog()
			if err != nil {
				errorCounter.WithLabelValues("clean_slow_log").Inc()
				a.logger.Error("failed to clean slow log", zap.Error(err))
			}

			err = a.cleanLog()
			if err != nil {
				errorCounter.WithLabelValues("clean_log").Inc()
				a.logger.Error("failed to clean log", zap.Error(err))
			}
		case <-ctx.Done():
			a.logger.Info("audit log server exiting")

			// Tidb assumed to be quit now and not open and writing the file.
			//
			// Avoid rotate and push log file ourself if kill the auditlog app directly.
			// etc:
			// while true; do k exec -it db-tidb-0 -c auditlog – /bin/sh -c "while true; do kill 1; done" ; sleep 1; done
			if producerPort := a.ProducerPort; producerPort != 0 {
				if canConn("localhost:"+strconv.Itoa(producerPort)) == nil && !a.skipCheckConn {
					a.logger.Error("tidb is still running, skip rotate and push log")
					return nil
				}
			}

			_, err := a.rotateIfNeed(a.LogDir, a.AuditLogFileName)
			if err != nil {
				a.logger.Error("failed to try rotate if need", zap.Error(err))
			}

			//nolint:contextcheck //try the best to push the log before quit
			_, a.pushCountInShutdown, err = a.pushLog(context.Background(), a.handler, 0 /*limitNum*/)
			if err != nil {
				a.logger.Error("failed to push log", zap.Error(err))
				return err
			}
			pushNumInShutdown.Set(float64(a.pushCountInShutdown))
			a.logger.Info("audit log server exited", zap.Int("pushCount", a.pushCountInShutdown))
			return nil
		}
	}
}

func (a *Server) Shutdown(ctx context.Context, cancel context.CancelFunc) {
	if a.GracefulStopWaitSeconds > 0 {
		a.logger.Info("wait before starting quit", zap.Int("seconds", a.GracefulStopWaitSeconds))
		time.Sleep(time.Second * time.Duration(a.GracefulStopWaitSeconds))
	}
	a.logger.Info("graceful wait end, start to quit")
	cancel()
}

func (a *Server) pushLog(ctx context.Context, handler LogHandler, limitNum int) (more bool, pushCount int, err error) {
	if a.LogDir == "" {
		pendingPushFiles.Set(0)
		a.logger.Debug("skip push: logDir is empty", zap.String("bucketURL", a.BucketURL))
		return false, 0, nil
	}

	logs, err := listLog(a.LogDir, a.AuditLogFileName)
	if err != nil {
		return false, 0, err
	}
	pendingPushFiles.Set(float64(len(logs)))
	if a.BucketURL == "" {
		// Do not try to push log if BucketURL is empty.
		// Otherwise, the log will be deleted directly as the default `NoPushHandler.Push` always returns nil.
		a.logger.Debug("skip push: bucketURL is empty", zap.String("logDir", a.LogDir))
		return false, 0, nil
	}
	if len(logs) == 0 {
		a.logger.Debug("no rotated log files to push",
			zap.String("logDir", a.LogDir),
			zap.String("auditLogFileName", a.AuditLogFileName))
		return false, 0, nil
	}

	u, err := url.Parse(a.BucketURL)
	if err != nil {
		return false, 0, errors.Annotatef(err, "invalid url %q", a.BucketURL)
	}

	prefix := u.Path
	prefix = strings.TrimLeft(prefix, "/")

	logsToProcess := logs
	if limitNum != 0 && len(logs) > limitNum {
		logsToProcess = logs[:limitNum]
	}
	if len(logsToProcess) == 1 {
		return a.pushSingleLog(ctx, handler, logsToProcess[0], prefix, len(logs) > len(logsToProcess))
	}
	return a.pushLogsConcurrently(ctx, handler, logsToProcess, prefix, len(logs) > len(logsToProcess))
}

// pushSingleLog handles pushing a single log file without worker pool overhead
func (a *Server) pushSingleLog(ctx context.Context, handler LogHandler, item logItem, prefix string, hasMore bool) (more bool, pushCount int, err error) {
	fname := filepath.Join(a.LogDir, item.name)

	if item.size > 0 { // if size is 0, just delete it directly
		start := time.Now()
		targetKey := genTargetKey(prefix, item.name, a.AuditLogFileName)
		err := handler.Push(ctx, fname, targetKey)
		if err != nil {
			pushLogCounter.WithLabelValues(LabelResultFail).Add(1)
			pushLogDuration.WithLabelValues(LabelResultFail).Observe(time.Since(start).Seconds())
			return false, 0, errors.Annotatef(err, "failed to push log %q", fname)
		}

		pushLogCounter.WithLabelValues(LabelResultSuccess).Add(1)
		pushLogDuration.WithLabelValues(LabelResultSuccess).Observe(time.Since(start).Seconds())
		a.logger.Info("push file success",
			zap.String("name", fname),
			zap.Int64("size", item.size),
			zap.Duration("takes", time.Since(start)),
		)
	}

	err = os.Remove(fname)
	if err != nil {
		return false, 0, errors.Annotatef(err, "failed to delete log %q", fname)
	}
	pendingPushFiles.Dec()

	a.logger.Info("delete file success", zap.String("name", fname))
	return hasMore, 1, nil
}

// pushLogsConcurrently handles pushing multiple log files concurrently using a worker pool
func (a *Server) pushLogsConcurrently(ctx context.Context, handler LogHandler, logs []logItem, prefix string, hasMore bool) (more bool, pushCount int, err error) {
	var (
		wg       sync.WaitGroup
		mu       sync.Mutex
		firstErr error
		count    int
	)

	concurrency := a.PushConcurrency
	if len(logs) < concurrency {
		concurrency = len(logs)
	} else if concurrency <= 0 {
		concurrency = defaultPushConcurrency
	}
	ch := make(chan struct{}, concurrency)

	for _, log := range logs {
		mu.Lock()
		if firstErr != nil {
			// break if an error has occurred
			// the ongoing goroutines will still finish
			mu.Unlock()
			break
		}
		mu.Unlock()

		wg.Add(1)
		ch <- struct{}{}

		go func(log logItem) {
			defer wg.Done()
			defer func() { <-ch }()

			_, pushCount, pushErr := a.pushSingleLog(ctx, handler, log, prefix, false)

			mu.Lock()
			if pushErr != nil {
				if firstErr == nil {
					firstErr = pushErr
				}
			} else {
				count += pushCount
			}
			mu.Unlock()
		}(log)
	}

	wg.Wait()

	mu.Lock()
	defer mu.Unlock()

	if firstErr != nil {
		return false, count, firstErr
	}

	return hasMore, count, nil
}

// after changing the log format between TEXT and JSON via SQL variable `tidb_audit_log_format`,
// we need to rotate the old log file to ensure it can be pushed to the remote storage.
// for example:
//  1. writing log in TEXT format, the log file is `tidb-audit.log`
//  2. change the log format to JSON via SQL variable `tidb_audit_log_format`
//  3. a new log file `tidb-audit.log.json` will be created, and the old log file `tidb-audit.log` will still be there.
//  4. we need to rotate the old log file `tidb-audit.log`
//     to ensure it can be pushed to the remote storage.
func (a *Server) rotateIfLogFormatChanged() {
	// This is a TiDB-specific feature, skip for other modes
	if Mode(a.Mode) != ModeTiDB {
		return
	}
	if a.BucketURL == "" || a.LogDir == "" {
		return
	}
	f1name := filepath.Join(a.LogDir, a.AuditLogFileName)
	stat1, err := os.Stat(f1name)
	if err != nil {
		if !stderrors.Is(err, os.ErrNotExist) {
			a.logger.Error("failed to stat audit log file", zap.String("name", f1name), zap.Error(err))
		}
		return
	}
	f2name := filepath.Join(a.LogDir, a.AuditLogFileName+".json")
	stat2, err := os.Stat(f2name)
	if err != nil {
		if !stderrors.Is(err, os.ErrNotExist) {
			a.logger.Error("failed to stat audit log file", zap.String("name", f2name), zap.Error(err))
		}
		return
	}

	if stat1.Size() == 0 && stat2.Size() == 0 {
		return
	}

	// check modified time of the two files to determine which one is newer.
	// add a difference of 2 seconds to avoid the case that the two files have the same modified time.
	// NOTE: we have a cooldown of audit setting change, so it should not change from TEXT to JSON and then back to TEXT
	// in a very short time, so we can safely use the modified time to determine which file is newer.
	var (
		emptyFilename  string
		rotateFilename string
	)
	//nolint:gocritic // if-else instead of switch is more readable here.
	if stat1.ModTime().After(stat2.ModTime().Add(2 * time.Second)) {
		if stat2.Size() == 0 {
			emptyFilename = f2name
		} else {
			rotateFilename = f2name
		}
	} else if stat2.ModTime().After(stat1.ModTime().Add(2 * time.Second)) {
		if stat1.Size() == 0 {
			emptyFilename = f1name
		} else {
			rotateFilename = f1name
		}
	} else {
		// if both files are modified at the same time, we do not rotate any of them.
		// wait for the next log write to the real new file to trigger the rotation.
		return
	}

	if emptyFilename != "" {
		// if the file is empty, we can just delete it.
		// if we rotate it, a new empty file will be created again.
		if err = os.Remove(emptyFilename); err != nil {
			a.logger.Error("failed to delete empty log file", zap.String("name", emptyFilename), zap.Error(err))
		} else {
			a.logger.Info("delete empty log file success", zap.String("name", emptyFilename))
		}
		return
	}
	if rotateFilename != "" {
		// if the file is not empty, we need to rotate it.
		// after rotating, it will be pushed to the remote storage.
		lg := lumberjack.Logger{
			Filename: rotateFilename,
		}
		defer lg.Close()
		if err := lg.Rotate(); err != nil {
			a.logger.Error("failed to rotate log file", zap.String("name", rotateFilename), zap.Error(err))
		} else {
			a.logger.Info("rotate log file success", zap.String("name", rotateFilename))
			// remove the new & empty file after rotating
			// so that it would not trigger another compare and rotate with the new format file
			lg.Close()
			if err := os.Remove(rotateFilename); err != nil {
				a.logger.Error("failed to delete rotated empty log file", zap.String("name", rotateFilename), zap.Error(err))
			} else {
				a.logger.Info("delete rotated empty log file success", zap.String("name", rotateFilename))
			}
		}
		return
	}
}

// MaxDays and MaxBackups not working for slow-query-logger in some versions
// if SlowLogMaxBackups is not 0, we will delete the slow log files to keep at most SlowLogMaxBackups
// ourself to avoid slow low take too much space.
// ref: https://github.com/pingcap/tidb/pull/27625
func (a *Server) cleanSlowlog() error {
	if Mode(a.Mode) == ModeAgent {
		// skip slow log cleaning in Agent mode
		return nil
	}
	if a.SlowLogMaxBackups == 0 || a.LogDir == "" {
		return nil
	}

	logs, err := listLog(a.LogDir, a.SlowLogFileName)
	if err != nil {
		return errors.Annotate(err, "failed to get list log")
	}

	if len(logs) <= a.SlowLogMaxBackups {
		return nil
	}

	logs = logs[:len(logs)-a.SlowLogMaxBackups]

	for _, log := range logs {
		fname := filepath.Join(a.LogDir, log.name)
		err := os.Remove(fname)
		if err != nil {
			return errors.Annotatef(err, "failed to delete file %q", fname)
		}
	}

	return nil
}

func (a *Server) cleanLog() error {
	if a.LogDir == "" {
		pendingPushFiles.Set(0)
		return nil
	}
	storageSize, err := getStorageSize(a.LogDir)
	if err != nil {
		return errors.Annotate(err, "failed to get storage size")
	}

	storageSizeCapacityBytes.Set(float64(storageSize.capacity))
	storageSizeAvailableBytes.Set(float64(storageSize.available))

	logs, err := listLog(a.LogDir, a.AuditLogFileName)
	if err != nil {
		return errors.Annotate(err, "failed to get list log")
	}
	pendingPushFiles.Set(float64(len(logs)))

	for _, log := range logs {
		if (1.0 - float64(storageSize.available)/float64(storageSize.capacity)) < a.CleanOldestLogPercent {
			return nil
		}

		fname := filepath.Join(a.LogDir, log.name)
		err := os.Remove(fname)
		if err != nil {
			return errors.Annotatef(err, "failed to delete file %q", fname)
		}
		pendingPushFiles.Dec()
		a.logger.Info("delete file to free space", zap.String("name", fname))
		deleteLogFileDueToDiskFullCounter.Add(1)

		storageSize.available += uint64(log.size)
	}

	return nil
}

func (a *Server) buildLogHandler() LogHandler {
	noPush := defaultLogHandler
	if a.BucketURL == "" {
		// do not try to push log if BucketURL is empty.
		return noPush
	}

	logger := a.logger.With(zap.String("url", a.BucketURL))

	u, err := url.Parse(a.BucketURL)
	if err != nil {
		logger.Error("wrong bucketURL", zap.Error(err))
	}

	switch u.Scheme {
	case "s3":
		h, err := NewS3Handler(a.Config)
		if err != nil {
			logger.Error("failed to create s3 handler", zap.Error(err))
			return noPush
		}

		return h
	case "gcs":
		h, err := NewGCSHandler(a.Config)
		if err != nil {
			logger.Error("failed to create gcs handler", zap.Error(err))
			return noPush
		}

		return h
	case "https": // azure blob url format https://{account}.blob.core.windows.net/{container}/{path}
		h, err := NewAZBlobHandler(a.Config)
		if err != nil {
			logger.Error("failed to create azblob handler", zap.Error(err))
			return noPush
		}

		return h
	case "oss": // AliCloud oss url format oss://{bucket}.{endpoint}/{path}
		h, err := NewOSSHandler(a.Config)
		if err != nil {
			logger.Error("failed to create oss handler", zap.Error(err))
			return noPush
		}

		return h
	default:
		logger.Error("unknown scheme in url", zap.Error(err))
		return noPush
	}
}

type logItem struct {
	name string
	size int64
}

// no ext example:
// slowlog slowlog-2022-02-07T13-09-09.140
// ext example:
// slowlog.log slowlog-2022-02-07T13-09-09.140.log
// audit log example:
// TEXT: tidb-audit-2025-08-01T09-42-10.442.log
// JSON: tidb-audit.log-2025-08-01T09-42-10.442.json
func isRotatedFile(name, checkName string) bool {
	if name == checkName || name+".json" == checkName {
		return false
	}

	nameExt := filepath.Ext(name)
	prefix := name[:len(name)-len(nameExt)]

	// roughly match by just checking having prefix.
	if strings.HasPrefix(checkName, prefix+"-") {
		return true
	}
	if strings.HasPrefix(checkName, prefix+".") {
		// check `.json` extension for JSON format.
		if strings.HasSuffix(checkName, ".json") {
			return true
		}
	}

	return false
}

// listLog return log file name sort by time ASC.
// NOTE: this function will not return the file which is not rotated.
// For example, if the log file is `tidb-audit.log`, it will not return it.
func listLog(dir, logName string) ([]logItem, error) {
	var logs []logItem

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

		name := info.Name()

		if info.IsDir() && path != dir {
			return filepath.SkipDir
		}

		if isRotatedFile(logName, name) {
			logs = append(logs, logItem{
				name: name,
				size: info.Size(),
			})
		}

		return nil
	})

	// For filepath.Walk, the files are walked in lexical order, so the log files is already sort by time ASC.
	return logs, err
}

type storageSize struct {
	capacity  uint64
	available uint64
}

func getStorageSize(dir string) (size storageSize, err error) {
	var stat unix.Statfs_t

	err = unix.Statfs(dir, &stat)
	if err != nil {
		return size, errors.Trace(err)
	}

	// When container is run in MacOS, `bsize` obtained by `statfs` syscall is not the fundamental block size,
	// but the `iosize` (optimal transfer block size) instead, it's usually 1024 times larger than the `bsize`.
	// for example `4096 * 1024`. To get the correct block size, we should use `frsize`. But `frsize` isn't
	// guaranteed to be supported everywhere, so we need to check whether it's supported before use it.
	// For more details, please refer to: https://github.com/docker/for-mac/issues/2136
	bSize := uint64(stat.Bsize)
	field := reflect.ValueOf(&stat).Elem().FieldByName("Frsize")
	if field.IsValid() {
		if field.Kind() == reflect.Uint64 {
			bSize = field.Uint()
		} else {
			bSize = uint64(field.Int())
		}
	}

	// Available blocks * size per block = available space in bytes
	size.available = stat.Bavail * bSize
	size.capacity = stat.Blocks * bSize

	return
}

func (a *Server) rotateIfNeed(dir, name string) (rotated bool, err error) {
	if dir == "" || name == "" {
		a.logger.Debug("skip rotation: dir or name is empty",
			zap.String("dir", dir),
			zap.String("name", name))
		return false, nil
	}
	fname := filepath.Join(dir, name)
	stat, err := os.Stat(fname)
	if err != nil && !stderrors.Is(err, os.ErrNotExist) {
		return false, err
	}

	actualName := name
	if (err != nil && stderrors.Is(err, os.ErrNotExist)) || stat.Size() == 0 {
		// try to rotate the file with `.json` suffix.
		actualName = name + ".json"
		fname = filepath.Join(dir, actualName)
		stat, err = os.Stat(fname)
		if err != nil {
			if stderrors.Is(err, os.ErrNotExist) {
				a.logger.Debug("log file not found, skip rotation",
					zap.String("tried", filepath.Join(dir, name)),
					zap.String("triedJson", fname))
				return false, nil
			}
			return false, err
		}
	}

	if stat.Size() == 0 {
		a.logger.Debug("log file is empty, skip rotation", zap.String("file", fname))
		return false, nil
	}

	// Use the configured rotation strategy
	rotator := NewRotator(RotationStrategy(a.RotationStrategy))
	backupName, err := rotator.Rotate(dir, actualName)
	if err != nil {
		return false, err
	}
	if backupName != "" {
		a.logger.Info("rotated log file",
			zap.String("file", fname),
			zap.String("backup", backupName),
			zap.String("strategy", a.RotationStrategy))
	}
	return true, nil
}

// defaultAgentSubPath is the default sub path for agent mode when SubPath is not configured.
const defaultAgentSubPath = "dataplane-agent"

// getApplicationPrefix returns the application prefix for bucket path generation.
// In tidb mode, uses built-in TiDB/Kong detection for backward compatibility.
func (a *Server) getApplicationPrefix(podName string) string {
	if podName == "" {
		return ""
	}

	// Default TiDB mode behavior: detect TiDB or Kong
	if strings.Contains(podName, "kong") {
		return "kong-"
	}
	return "tidb-"
}

// appendSuffix appends path suffix to the bucket URL based on application configuration.
// In agent mode: uses SubPath directly.
// In tidb mode: appends prefix (tidb-/kong-) + pod suffix.
func (a *Server) appendSuffix(burl, podName string) (string, error) {
	if burl == "" {
		return burl, nil
	}

	u, err := url.Parse(burl)
	if err != nil {
		return burl, err
	}

	if podName == "" {
		return burl, nil
	}

	// Agent mode: use SubPath directly without suffix
	if Mode(a.Mode) == ModeAgent {
		subPath := a.SubPath
		if subPath == "" {
			subPath = defaultAgentSubPath
		}
		u.Path = filepath.Join(u.Path, subPath)
		return u.String(), nil
	}

	// TiDB mode: append prefix + suffix
	suffix := getSuffixFromPodName(podName)
	prefix := a.getApplicationPrefix(podName)

	if suffix != "" && prefix != "" {
		u.Path = filepath.Join(u.Path, prefix+suffix)
		return u.String(), nil
	}

	return burl, nil
}

// The Pod name format in TiDB Operator v2 is something like: "db-1949758074227003392-tidb-p6u2xw".
// We can get the suffix from the Pod name and use it to append to the bucket URL.
// NOTE: the second part (`1949758074227003392` here) is TidbGroup name,
// if we want to push the log to different TidbGroup, we should configure different bucket URL.
func getSuffixFromPodName(name string) string {
	ar := strings.Split(name, "-")
	if len(ar) <= 1 {
		return ""
	}
	return ar[len(ar)-1]
}

func genTargetKey(prefix, filename, baseFilename string) string {
	if !strings.Contains(prefix, dateTmplPlaceholder) {
		return filepath.Join(prefix, filename)
	}

	// get the date from the name, the name format is like:
	// TEXT: tidb-audit-2025-08-01T09-42-10.442.log
	// JSON: tidb-audit.log-2025-08-01T09-42-10.442.json
	// trim the prefix and suffix based on the basename.
	// for example, if the basename is `tidb-audit.log`, we will trim
	// the prefix `tidb-audit-` or `tidb-audit.log-` and suffix `.log` or `.json`.
	// so the date will be `2025-08-01T09-42-10.442`.
	// then we can replace the `{{.date}}` placeholder with the date.
	basename := baseFilename[:len(baseFilename)-len(filepath.Ext(baseFilename))]
	var timeStr string
	//nolint:gocritic // if-else instead of switch is more readable here.
	if strings.HasSuffix(filename, ".log") {
		timeStr = strings.TrimPrefix(filename, basename+"-")
		timeStr = strings.TrimSuffix(timeStr, ".log")
	} else if strings.HasSuffix(filename, ".json") {
		timeStr = strings.TrimPrefix(filename, basename+".log-")
		timeStr = strings.TrimSuffix(timeStr, ".json")
	} else {
		// if the file is not in the expected format, just return the filename.
		return filepath.Join(prefix, filename)
	}
	// get the date part from the time string.
	// extract the first 10 characters as the date in the format `YYYY-MM-DD`.
	if len(timeStr) < 10 {
		return filepath.Join(prefix, filename)
	}
	date := timeStr[:10]
	// replace the `{{.date}}` placeholder with the date.
	// NOTE: we need convert 'YYYY-MM-DD' to'YYYYMMDD' format.
	date = strings.ReplaceAll(date, "-", "")
	prefix = strings.ReplaceAll(prefix, dateTmplPlaceholder, date)
	return filepath.Join(prefix, filename)
}

func canConn(addr string) error {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		return err
	}

	conn.Close()
	return nil
}