// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package binlog

import (
	"path"
	"strconv"
	"time"

	gmysql "github.com/go-mysql-org/go-mysql/mysql"
	"github.com/pingcap/tidb/dumpling/export"
	"github.com/pingcap/tiflow/dm/pkg/conn"
	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
	"github.com/pingcap/tiflow/dm/pkg/terror"
	"github.com/pingcap/tiflow/dm/pkg/utils"
)

// in MySQL, we can set `max_binlog_size` to control the max size of a binlog file.
// but this is not absolute:
// > A transaction is written in one chunk to the binary log, so it is never split between several binary logs.
// > Therefore, if you have big transactions, you might see binary log files larger than max_binlog_size.
// ref: https://dev.mysql.com/doc/refman/5.7/en/replication-options-binary-log.html#sysvar_max_binlog_size
// The max value of `max_binlog_size` is 1073741824 (1GB)
// but the actual file size still can be larger, and it may exceed the range of an uint32
// so, if we use go-mysql.Position(with uint32 Pos) to store the binlog size, it may become out of range.
// ps, use go-mysql.Position to store a position of binlog event (position of the next event) is enough.
type binlogSize struct {
	name string
	size int64
}

// FileSizes is a list of binlog filename and size.
type FileSizes []binlogSize

// GetBinaryLogs returns binlog filename and size of upstream.
func GetBinaryLogs(ctx *tcontext.Context, db *conn.BaseDB) (FileSizes, error) {
	query := "SHOW BINARY LOGS"
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		return nil, terror.DBErrorAdapt(err, db.Scope, terror.ErrDBDriverError)
	}
	defer func() {
		_ = rows.Close()
		_ = rows.Err()
	}()
	files := make([]binlogSize, 0, 10)
	var rowsResult [][]string
	rowsResult, err = export.GetSpecifiedColumnValuesAndClose(rows, "Log_name", "File_size")
	if err != nil {
		return nil, terror.DBErrorAdapt(err, db.Scope, terror.ErrDBDriverError)
	}
	for _, rowResult := range rowsResult {
		pos, err := strconv.ParseInt(rowResult[1], 10, 64)
		if err != nil {
			return nil, terror.DBErrorAdapt(err, db.Scope, terror.ErrDBDriverError)
		}
		files = append(files, binlogSize{name: rowResult[0], size: pos})
	}
	return files, nil
}

// After returns the total size of binlog after `fromFile` in FileSizes.
func (b FileSizes) After(fromFile gmysql.Position) int64 {
	var total int64
	for _, file := range b {
		switch gmysql.CompareBinlogFileName(file.name, fromFile.Name) {
		case -1:
			continue
		case 1:
			total += file.size
		case 0:
			if file.size > int64(fromFile.Pos) {
				total += file.size - int64(fromFile.Pos)
			}
		}
	}

	return total
}

func GetLocalBinaryLogs(dir string) (FileSizes, error) {
	fileNames, err := ReadSortedBinlogFromDir(dir)
	if err != nil {
		return nil, err
	}
	files := make([]binlogSize, 0, len(fileNames))
	for _, fileName := range fileNames {
		size, err := utils.GetFileSize(path.Join(dir, fileName))
		if err != nil {
			return nil, err
		}
		files = append(files, binlogSize{name: fileName, size: size})
	}
	return files, nil
}

// SourceStatus collects all information of upstream.
type SourceStatus struct {
	Location   Location
	Binlogs    FileSizes
	UpdateTime time.Time
}

func GetSourceStatus(tctx *tcontext.Context, db *conn.BaseDB, flavor string) (*SourceStatus, error) {
	ret := &SourceStatus{}
	ctx, cancel := tctx.WithTimeout(conn.DefaultDBTimeout)
	defer cancel()
	pos, gtidSet, err := conn.GetPosAndGs(ctx, db, flavor)
	if err != nil {
		return nil, err
	}
	ret.Location = NewLocation(pos, gtidSet)
	ctx2, cancel2 := tctx.WithTimeout(conn.DefaultDBTimeout)
	defer cancel2()
	binlogs, err := GetBinaryLogs(ctx2, db)
	if err != nil {
		return nil, err
	}
	ret.Binlogs = binlogs

	ret.UpdateTime = time.Now()
	return ret, nil
}