package import_svc

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/parser"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/format"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	v1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/grpc/grpcli"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	ErrColumnsCountNotMatch   = "There are %d columns in the source data file and %d columns in the target table."
	ErrColumnsMissingInFile   = "The following columns exist in the target table, but not in the source file: %s."
	ErrColumnsMissingInTarget = "The following columns do not exist in the target table, but exist in the source file: %s."
)

func generateCreateDatabaseSQL(dbName string) string {
	return fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`;",
		utils.EscapeBackticks(dbName),
	)
}

func generateCreateTableSQL(
	logger *zap.Logger,
	dbName string,
	tableName string,
	columns []*entity.ColumnInfo,
	pkCols []string,
) (string, bizerr.BizError) {
	colDefStrs := make([]string, 0, len(columns))
	colMap := make(map[string]struct{})
	for i, colInfo := range columns {
		if colInfo.ColumnName == "" {
			return "", bizerr.New(bizerr.ErrCodeParams, "The %s column name is empty.", humanize.Ordinal(i+1))
		}
		colDefStrs = append(
			colDefStrs,
			fmt.Sprintf("`%s` %s", utils.EscapeBackticks(colInfo.ColumnName), colInfo.ColumnType),
		)
		colMap[strings.ToLower(colInfo.ColumnName)] = struct{}{}
	}

	// check whether pkCol is in the columns
	for i, pkCol := range pkCols {
		// the column name is case-insensitive
		if _, ok := colMap[strings.ToLower(pkCol)]; !ok {
			return "", bizerr.New(
				bizerr.ErrCodeParams,
				"Primary key column '%s' cannot be found in the table's column list.",
				pkCol,
			)
		}

		pkCols[i] = fmt.Sprintf("`%s`", utils.EscapeBackticks(pkCol))
	}
	var pkStr string
	if len(pkCols) > 0 {
		pkStr = fmt.Sprintf(", PRIMARY KEY( %s ) CLUSTERED", strings.Join(pkCols, ", "))
	}
	rawCreateTableStmtStr := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s`.`%s` ( %s %s );",
		utils.EscapeBackticks(dbName),
		utils.EscapeBackticks(tableName),
		strings.Join(colDefStrs, ", "),
		pkStr,
	)
	createStmtAST, err := parser.New().ParseOneStmt(rawCreateTableStmtStr, "", "")
	if err != nil {
		logger.Error("parse create statement error", zap.Error(err))
		return "", bizerr.New(
			bizerr.ErrCodeParams,
			"The table cannot be created due to invalid table name, column names or column types. Please check these settings.",
		)
	}
	var sb strings.Builder
	if err := createStmtAST.Restore(format.NewRestoreCtx(format.DefaultRestoreFlags, &sb)); err != nil {
		logger.Error("restore AST into SQL string error", zap.Error(err))
		return "", bizerr.New(errorcode.ErrCodeInternalError, "failed to generate create table SQL")
	}
	return sb.String(), nil
}

func convertStatus(status string) constant.ImportStatus {
	switch status {
	case string(v1.ImportCompleted):
		return constant.ImportStatusCompleted
	case string(v1.ImportFailed):
		return constant.ImportStatusFailed
	case string(v1.ImportPrepare), "":
		return constant.ImportStatusPreparing
	case string(v1.ImportRunning):
		return constant.ImportStatusImporting
	default:
		return constant.ImportStatusUnknown
	}
}

// batchLoadImportClusterInfo loads a list of import tasks' cluster information in a single API call.
func batchLoadImportClusterInfo(ctx context.Context, impTasks []*entity.ImportTask) ([]*entity.ImportTask, error) {
	if len(impTasks) == 0 {
		// avoid listing clusters with no cluster ID specified, in this case, it will iterate all the clusters
		return nil, nil
	}
	clusterIDs := make([]uint64, len(impTasks))
	for i, imp := range impTasks {
		clusterID, err := strconv.ParseUint(imp.ClusterID, 10, 64)
		if err != nil {
			return nil, err
		}
		clusterIDs[i] = clusterID
	}
	listResp, err := grpcli.ListClusters(ctx, &tidbmgmtservicev1.ListClustersRequest{
		Filter: &tidbmgmtservicev1.ClusterFilter{
			ClusterIds:  clusterIDs,
			View:        tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
			ClusterType: tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS.Enum(),
		},
	})
	if err != nil {
		return nil, errors.Annotate(err, "list clusters by ID error")
	}
	log.Info(ctx, "list clusters by ID", zap.Any("listResp", listResp))
	clusterMap := make(map[uint64]*tidbmgmtservicev1.Cluster)
	for _, cluster := range listResp.Clusters {
		clusterMap[cluster.GetClusterId()] = cluster
	}
	var missingClusterImportIDs []string
	var imps []*entity.ImportTask
	for _, imp := range impTasks {
		clusterID, err := strconv.ParseUint(imp.ClusterID, 10, 64)
		if err != nil {
			return nil, err
		}
		cluster, ok := clusterMap[clusterID]
		if !ok {
			missingClusterImportIDs = append(missingClusterImportIDs, imp.ID)
			continue
		}
		imp.Cluster = cluster
		imps = append(imps, imp)
	}
	if len(missingClusterImportIDs) > 0 {
		return imps, errors.Errorf("cannot find cluster information for some import: %v", missingClusterImportIDs)
	}
	return imps, nil
}

type showCreateTableResult struct {
	Table       [][]byte `json:"Table"`
	CreateTable [][]byte `json:"Create Table"`
}

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
}

func convertToTableResult(tblPhase v1.TableImportPhaseType) constant.ImportTableResult {
	switch tblPhase {
	case v1.TableImportPhaseImported, v1.TableImportPhaseComplete:
		return constant.ImportTableResultSuccess
	case v1.TableImportPhaseFailed:
		return constant.ImportTableResultError
	default:
		return ""
	}
}

func ValidateUpload(ctx context.Context, upload *entity.Upload, clusterID uint64) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", clusterID))

	if upload.Status != constant.UploadStatusCompleted {
		return bizerr.New(bizerr.ErrCodeParams, "The upload status is not completed")
	}
	if upload.CreatedAt.Add(constant.UploadExpireTime).Before(time.Now()) {
		return bizerr.New(bizerr.ErrCodeParams, "The upload file has expired")
	}
	if upload.ClusterID != clusterID {
		logger.Error(
			"The upload file cluster id is inconsistent",
			zap.Uint64("upload_cluster_id", upload.ClusterID),
			zap.Uint64("req_cluster_id", clusterID),
		)
		return bizerr.New(
			bizerr.ErrCodeRecordNotFound,
			"The upload file cluster id is inconsistent with create import request cluster id",
		)
	}
	return nil
}

func ParseColumns(table *agent.Table) []*entity.ColumnInfo {
	var columns []*entity.ColumnInfo
	for i := 0; i < len(table.RowHeaders); i++ {
		colName := table.RowHeaders[i]
		if colName == "" {
			colName = fmt.Sprintf("column_%d", i+1)
		}
		columns = append(columns, &entity.ColumnInfo{
			ColumnName: colName,
			ColumnType: table.RowDataTypes[i],
		})
	}
	return columns
}

func ValidateColumns(fileColumns, targetColumns []*entity.ColumnInfo) bizerr.BizError {
	fileColsLen := len(fileColumns)
	targetColsLen := len(targetColumns)

	if targetColsLen == 0 {
		return nil
	}
	if fileColsLen != targetColsLen {
		return bizerr.New(bizerr.ErrCodeParams, ErrColumnsCountNotMatch, fileColsLen, targetColsLen)
	}

	tmp := make(map[string]struct{}, fileColsLen)
	for _, c := range fileColumns {
		tmp[c.ColumnName] = struct{}{}
	}

	var missLocalFile, missTarget []string
	for _, c := range targetColumns {
		if _, ok := tmp[c.ColumnName]; !ok {
			missLocalFile = append(missLocalFile, c.ColumnName)
		} else {
			delete(tmp, c.ColumnName)
		}
	}

	for cname := range tmp {
		missTarget = append(missTarget, cname)
	}

	// since map is unordered, sort the result
	sort.Slice(missTarget, func(i, j int) bool { return missTarget[i] < missTarget[j] })

	errMsg := ""
	if len(missLocalFile) != 0 {
		errMsg += fmt.Sprintf(ErrColumnsMissingInFile, strings.Join(missLocalFile, ","))
	}
	if len(missTarget) != 0 {
		if errMsg != "" {
			errMsg += " \n"
		}
		errMsg += fmt.Sprintf(ErrColumnsMissingInTarget, strings.Join(missTarget, ","))
	}

	if errMsg != "" {
		return bizerr.New(bizerr.ErrCodeParams, "%s", errMsg)
	}
	return nil
}