package pimport

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"sort"
	"strings"
	"time"

	"github.com/docker/go-units"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/charset"
	"github.com/pingcap/tidb/pkg/parser/mysql"
	ptypes "github.com/pingcap/tidb/pkg/parser/types"
	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
	"github.com/pingcap/tidb/pkg/types"
	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"go.uber.org/zap"
)

func (c *Model) GenCloudAdminPassword(
	ctx context.Context,
	logger *zap.Logger,
	orgId, projectId, clusterId uint64,
) (string, error) {
	var err error
	confidentialInfo, err := grpcli.GetClusterConfidentialInfoByID(ctx, clusterId)
	if err != nil {
		return "", errors.Annotate(err, "get cluster's confidential information by ID error")
	}

	encryptedPwd := confidentialInfo.GetEncryptedCloudAdminPasswd()

	if len(encryptedPwd) == 0 {
		logger.Error("the cloud admin password is empty")
		return "", errors.New("the cloud admin password is empty")
	}
	password, err := c.DecryptBase64(encryptedPwd)
	if err != nil {
		logger.Error("failed to decrypt", zap.Error(err))
		return "", errors.Annotate(err, "decrypt password error")
	}
	return password, nil
}

// inferDataType infer the type by a datum and its string
// return the type and priority.
func inferDataType(strDatum types.Datum, str string) *ptypes.FieldType {
	var (
		ft  *ptypes.FieldType
		err error
		sc  = stmtctx.NewStmtCtxWithTimeZone(time.Local)
	)

	contain := func(s string, ru rune) bool {
		for _, ch := range s {
			if ch == ru {
				return true
			}
		}
		return false
	}

	// bigint
	ft = ptypes.NewFieldType(mysql.TypeLonglong)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil && !contain(str, '.') {
		return ft
	}

	// double
	ft = ptypes.NewFieldType(mysql.TypeDouble)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil {
		return ft
	}

	// date(0000-00-00)
	ft = ptypes.NewFieldType(mysql.TypeDate)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil && !contain(str, ':') {
		return ft
	}

	// time(00:00:00)
	ft = ptypes.NewFieldType(mysql.TypeDuration)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil && !contain(str, '-') {
		return ft
	}

	// datetime(0000-00-00 00:00:00)
	ft = ptypes.NewFieldType(mysql.TypeDatetime)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil {
		return ft
	}

	// json
	ft = ptypes.NewFieldType(mysql.TypeJSON)
	_, err = strDatum.ConvertTo(sc.TypeCtx(), ft)
	if err == nil {
		return ft
	}

	// varchar(512)
	ft = ptypes.NewFieldType(mysql.TypeVarchar)
	ft.SetFlen(512)
	return ft
}

func GetDefaultDataType() []string {
	dataTypes := []*types.FieldType{ // corresponding string format
		ptypes.NewFieldType(mysql.TypeBit),               // bit
		ptypes.NewFieldType(mysql.TypeBlob),              // text
		ptypes.NewFieldType(mysql.TypeDate),              // date
		ptypes.NewFieldType(mysql.TypeDatetime),          // datetime
		ptypes.NewFieldType(mysql.TypeFloat),             // float
		ptypes.NewFieldType(mysql.TypeDouble),            // double
		ptypes.NewFieldType(mysql.TypeInt24),             // mediumint
		ptypes.NewFieldType(mysql.TypeJSON),              // json
		ptypes.NewFieldType(mysql.TypeLong),              // int
		ptypes.NewFieldType(mysql.TypeLonglong),          // bigint
		ptypes.NewFieldType(mysql.TypeLongBlob),          // longtext
		ptypes.NewFieldType(mysql.TypeMediumBlob),        // mediumtext
		ptypes.NewFieldType(mysql.TypeShort),             // smallint
		ptypes.NewFieldType(mysql.TypeDuration),          // time
		ptypes.NewFieldType(mysql.TypeTimestamp),         // timestamp
		ptypes.NewFieldType(mysql.TypeTiny),              // tinyint
		ptypes.NewFieldType(mysql.TypeTinyBlob),          // tinytext
		ptypes.NewFieldType(mysql.TypeTiDBVectorFloat32), // vector
	}

	// year(4)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeYear))
	dataTypes[len(dataTypes)-1].SetFlen(4)

	// boolean:tinyint(1)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeTiny))
	dataTypes[len(dataTypes)-1].SetFlen(1)

	// decimal[32[,4]]
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeNewDecimal))
	dataTypes[len(dataTypes)-1].SetFlen(32)
	dataTypes[len(dataTypes)-1].SetDecimal(4)

	// char(32)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeString))
	dataTypes[len(dataTypes)-1].SetFlen(32)
	// varchar(64)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeVarchar))
	dataTypes[len(dataTypes)-1].SetFlen(64)
	// varchar(256)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeVarchar))
	dataTypes[len(dataTypes)-1].SetFlen(256)
	// varchar(512)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeVarchar))
	dataTypes[len(dataTypes)-1].SetFlen(512)

	// binary(32)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeString))
	dataTypes[len(dataTypes)-1].SetFlen(32)
	dataTypes[len(dataTypes)-1].SetCharset(charset.CharsetBin)
	// varbinary(64)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeVarchar))
	dataTypes[len(dataTypes)-1].SetFlen(64)
	dataTypes[len(dataTypes)-1].SetCharset(charset.CharsetBin)
	// varbinary(256)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeVarchar))
	dataTypes[len(dataTypes)-1].SetFlen(256)
	dataTypes[len(dataTypes)-1].SetCharset(charset.CharsetBin)
	// varbinary(512)
	dataTypes = append(dataTypes, ptypes.NewFieldType(mysql.TypeVarchar))
	dataTypes[len(dataTypes)-1].SetFlen(512)
	dataTypes[len(dataTypes)-1].SetCharset(charset.CharsetBin)

	res := make([]string, len(dataTypes))
	for i := range dataTypes {
		res[i] = strings.ToUpper(dataTypes[i].CompactStr())
	}
	sort.StringSlice(res).Sort()
	return res
}

func columnIsPkuk(col *ast.ColumnDef, tableConstraints []*ast.Constraint) bool {
	if col == nil {
		return false
	}
	for _, opt := range col.Options {
		if opt.Tp == ast.ColumnOptionPrimaryKey || opt.Tp == ast.ColumnOptionUniqKey {
			return true
		}
	}
	for _, tblConstraint := range tableConstraints {
		if tblConstraint.Tp != ast.ConstraintPrimaryKey &&
			tblConstraint.Tp != ast.ConstraintUniqKey &&
			tblConstraint.Tp != ast.ConstraintUniq {
			continue
		}
		for _, key := range tblConstraint.Keys {
			if key.Column.Name.L == col.Name.Name.L {
				return true
			}
		}
	}
	return false
}

func columnIsAutoGen(col *ast.ColumnDef) bool {
	if col == nil {
		return false
	}
	for _, opt := range col.Options {
		if opt.Tp == ast.ColumnOptionAutoIncrement || opt.Tp == ast.ColumnOptionAutoRandom {
			return true
		}
	}
	return false
}

// Assuming it takes 50 seconds to import a 50MiB file,
// estimate the progress of the import by calculating
// the time difference between the start time of the
// import and the present time.
func estimateProgress(startTime time.Time, fileSize uint32) (estimatedCompletedPercent uint32) {
	elapsed := uint32(time.Since(startTime).Seconds())
	// 50MiB per 50 sec
	maxTime := uint32(50)
	maxFileSize := uint32(50 * units.MiB)
	estimatedSpeed := maxFileSize / maxTime

	// if the file size is unknown, use the estimated speed
	if fileSize == 0 {
		fileSize = maxFileSize
	}

	estimatedCompletedElapsed := fileSize / estimatedSpeed
	if estimatedCompletedElapsed < 1 {
		return 99
	}
	estimatedCompletedPercent = elapsed * 100 / estimatedCompletedElapsed
	if estimatedCompletedPercent >= 100 {
		return 99
	}
	return estimatedCompletedPercent
}

func getLineTerminator(reader io.Reader) (string, error) {
	bufReader := bufio.NewReader(reader)
	line, err := bufReader.ReadString('\n')
	if err != nil {
		// for one line file does not contain terminator, we do not report
		// error to user, just return "\n" as default.
		if err == io.EOF {
			return "\\n", nil
		}
		return "", err
	}

	if len(line) > 0 && line[len(line)-1] == '\n' {
		if len(line) > 1 && line[len(line)-2] == '\r' {
			return "\\r\\n", nil
		}
		return "\\n", nil
	}

	return "", nil
}

func GenTableID(schema, table string) string {
	return fmt.Sprintf("`%s`.`%s`", EscapeName(schema), EscapeName(table))
}

// EscapeName replaces all "`" in name with double "`"
func EscapeName(name string) string {
	return strings.Replace(name, "`", "``", -1)
}

// reorderDataColumns sort data by the original data from source and
// the column namp map(columnName -> original data index)
// This function is same as pkg/agent/lightning/preiview.go:sortData which will be removed later.
func reorderDataColumns(rowData []*agentpb.Row, sourceColNames, targetColNames []string) []*agentpb.Row {
	targetColNamesMap := make(map[string]struct{})
	for _, col := range targetColNames {
		targetColNamesMap[strings.ToLower(col)] = struct{}{}
	}

	rowValues := make([]*agentpb.Row, 0)

	if len(rowData) == 0 || len(rowData[0].ColumnValues) == 0 {
		return rowValues
	}
	// colsMap: columnName -> source column's index
	colsMap := make(map[string]int)
	for col := range targetColNamesMap {
		// the `col` from target table does not exist in data/schema file
		colsMap[col] = -1
	}

	if len(sourceColNames) > 0 {
		for i, col := range sourceColNames {
			col = strings.ToLower(col)
			if _, ok := targetColNamesMap[col]; ok {
				// the `colName` from target table is column i in data/schema file
				colsMap[col] = i
			}
		}
	} else {
		// If no column in source, we just show the rowData directly
		for i, col := range targetColNames {
			col = strings.ToLower(col)
			if i >= len(rowData[0].ColumnValues) {
				break
			}
			colsMap[col] = i
		}
	}

	for _, oneRow := range rowData {
		strRow := make([]string, 0, len(targetColNames))
		for _, col := range targetColNames {
			col = strings.ToLower(col)
			if colsMap[col] == -1 {
				strRow = append(strRow, "")
			} else {
				strRow = append(strRow, oneRow.ColumnValues[colsMap[col]])
			}
		}
		rowValues = append(rowValues, &agentpb.Row{ColumnValues: strRow})
	}
	return rowValues
}