package server

import (
	"context"
	stderrors "errors"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	"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"
	"gopkg.in/yaml.v2"
)

var defaultCheckLogDuration = time.Second * 5
var defaultCleanLogDuration = time.Second * 10
var defaultLogHandler LogHandler = &NoPushHandler{}

const auditLogFileName = "tidb-audit.log"
const slowLogFileName = "slowlog"

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

	Config

	handler LogHandler

	skipCheckConn bool
}

type Config struct {
	LogDir string `yaml:"logDir"`
	// An empty BucketURL will delete the rotated log directly.
	BucketURL  string `yaml:"bucketURL"`
	Region     string `yaml:"region"`
	RoleARN    string `yaml:"roleARN"`
	ExternalID string `yaml:"externalID"`

	SlowLogMaxBackups int `yaml:"slowLogMaxBackups"`
	// if disk is cleanOldestLogPercent full, we clean the oldest audit log files directly.
	// Should in  range [0.0, 1.0]
	CleanOldestLogPercent   float64 `yaml:"cleanOldestLogPercent"`
	GracefulStopWaitSeconds int     `yaml:"gracefulStopWaitSeconds"`

	// This DO NOT support reload config now.
	// Zero will not serve metrics
	MetricsPort int `yaml:"metricsPort"`
}

func (c *Config) ToYAML() ([]byte, error) {
	data, err := yaml.Marshal(c)
	return data, err
}

// 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("slowLogMaxBackups", 1)
	vp.SetDefault("cleanOldestLogPercent", 0.8)

	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)
	err := http.ListenAndServe(":"+strconv.Itoa(port), nil)

	return err
}

// 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
	}

	a.logger.Info("reload config", zap.Reflect("config", tmp))
	if withOrd, err := appendOrdinals(tmp.BucketURL, os.Getenv("POD_NAME")); err == nil {
		a.logger.Info("append pod orinal", zap.String("url", withOrd), zap.String("origin", tmp.BucketURL))
		tmp.BucketURL = withOrd
	}

	a.Config = tmp

	// Do not replace if we 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.Config.MetricsPort != 0 {
		go func() {
			err := a.serveMetrics(a.Config.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(defaultCheckLogDuration)
	cleanLogTick := time.NewTicker(defaultCleanLogDuration)

	pushFilesPerTime := 3

	for {
		select {
		case <-configChangeCh:
			a.loadConfig()
		case <-handlerCh:
			more, err := a.pushLog(ctx, a.handler, pushFilesPerTime)
			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():
			if a.GracefulStopWaitSeconds > 0 {
				a.logger.Info("wait before starting quit", zap.Int("seconds", a.GracefulStopWaitSeconds))
				time.Sleep(time.Second * time.Duration(a.GracefulStopWaitSeconds))
			}

			// 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 canConn("localhost:4000") == nil && !a.skipCheckConn {
				return nil
			}

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

			_, err = a.pushLog(context.Background(), a.handler, 0 /*limitNum*/)
			if err != nil {
				a.logger.Error("failed to push log", zap.Error(err))
				return err
			}
			return nil
		}
	}
}

func (a *Server) pushLog(ctx context.Context, handler LogHandler, limitNum int) (more bool, err error) {
	logs, err := listLog(a.LogDir, auditLogFileName)
	if err != nil {
		return false, err
	}

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

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

	for _, log := range logs {
		if limitNum != 0 && pushCount >= limitNum {
			return true, nil
		}

		start := time.Now()
		fname := filepath.Join(a.LogDir, log.name)
		targetKey := filepath.Join(prefix, log.name)
		err := handler.Push(ctx, fname, targetKey)
		if err != nil {
			pushLogDuration.WithLabelValues(LabelResultFail).Observe(time.Since(start).Seconds())
			return false, errors.Annotatef(err, "failed to push log %q", fname)
		}

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

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

		a.logger.Info("delete file success", zap.String("name", fname))

		pushCount++
	}

	return false, nil
}

// 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 a.SlowLogMaxBackups == 0 {
		return nil
	}

	logs, err := listLog(a.LogDir, 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
}

const checkFileName = "test-permission"

// Precheck try push one file to remote storage, return pass as false and the reason
// from log handler if fails to push.
func (a *Server) Precheck(ctx context.Context) (pass bool, reason string, err error) {
	file, err := os.CreateTemp("", "audit-log-precheck.*.temp")
	if err != nil {
		return false, "", errors.Annotate(err, "failed to create temp file")
	}
	defer os.Remove(file.Name())

	err = file.Close()
	if err != nil {
		return false, "", errors.Annotate(err, "failed to close file")
	}

	u, err := url.Parse(a.BucketURL)
	if err != nil {
		return false, "", err
	}

	prefix := u.Path
	prefix = strings.TrimLeft(prefix, "/")
	targetKey := filepath.Join(prefix, checkFileName)

	err = a.handler.Push(ctx, file.Name(), targetKey)
	if err != nil {
		return false, err.Error(), nil
	}

	return true, "", nil
}

func (a *Server) cleanLog() error {
	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, auditLogFileName)
	if err != nil {
		return errors.Annotate(err, "failed to get list log")
	}

	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)
		}
		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 == "" {
		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
	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
func isRotatedFile(name string, checkName string) bool {
	if name == 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
	}

	return false
}

// listLog return log file name sort by time ASC.
func listLog(dir string, 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 rotateIfNeed(dir string, name string) error {
	fname := filepath.Join(dir, name)
	stat, err := os.Stat(fname)
	if err != nil {
		if stderrors.Is(err, os.ErrNotExist) {
			return nil
		}
	}

	if stat.Size() == 0 {
		return nil
	}

	lg := lumberjack.Logger{
		Filename: fname,
	}

	err = lg.Rotate()
	return err
}

func appendOrdinals(burl string, 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
	}

	if podName != "" {
		ord := getOrdinalsFromPodName(podName)
		if ord != "" {
			u.Path = filepath.Join(u.Path, "tidb-"+ord)
			return u.String(), nil
		}
	}

	return burl, nil
}

func getOrdinalsFromPodName(name string) string {
	ar := strings.Split(name, "-")
	if len(ar) <= 1 {
		return ""
	}

	str := ar[len(ar)-1]
	_, err := strconv.Atoi(str)
	if err != nil {
		return ""
	}

	return str
}

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

	conn.Close()
	return nil
}