package grpc

import (
	"context"
	"database/sql"
	"fmt"
	"strings"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap/errors"
	plog "github.com/pingcap/log"
	"github.com/pingcap/tidb/pkg/errno"
	"github.com/pingcap/tidb/pkg/lightning/mydump"

	// "github.com/pingcap/tidb/pkg/parser/model"
	"github.com/pingcap/tidb/pkg/meta/model"
	pmysql "github.com/pingcap/tidb/pkg/parser/mysql"
	"github.com/pingcap/tidb/pkg/util/dbutil"
	"github.com/pingcap/tidb/pkg/util/dbutil/dbutiltest"
	fileparser "github.com/tidbcloud/dataflow-service/pkg/agent/file-parser"
	importcfg "github.com/tidbcloud/dataflow-service/pkg/import-config"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func checkTableEmpty(ctx context.Context, db *sql.DB, schema, table string) error {
	tableName := dbutil.TableName(schema, table)

	var exists int
	// same as https://github.com/pingcap/tidb/blob/9735568a3833719330c6efb787230a0a130ea72a/br/pkg/lightning/importer/get_pre_info.go#L195
	err := db.QueryRowContext(ctx, fmt.Sprintf("SELECT 1 FROM %s USE INDEX() LIMIT 1", tableName)).Scan(&exists)
	if err != nil {
		if mysqlErr, ok := errors.Cause(err).(*mysql.MySQLError); ok && mysqlErr.Number == errno.ErrNoSuchTable {
			return nil
		}
		if errors.ErrorEqual(err, sql.ErrNoRows) {
			return nil
		}
		return errors.Trace(err)
	}

	return status.Error(codes.PermissionDenied, fmt.Sprintf("table(%s) is not empty, please truncate table and retry it.", tableName))
}

type warningTable struct {
	schema string
	table  string
	cols   []string
}

// getTableInfo returns table information.
func getTableInfo(ctx context.Context, db dbutil.QueryExecutor, schemaName string, tableName string) (*model.TableInfo, error) {
	createTableSQL, err := dbutil.GetCreateTableSQL(ctx, db, schemaName, tableName)
	if err != nil {
		return nil, errors.Trace(err)
	}

	parser2, err := dbutil.GetParserForDB(ctx, db)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return dbutiltest.GetTableInfoBySQL(createTableSQL, parser2)
}

func checkColumnCompatitable(ctx context.Context, db *sql.DB, schema, table, sourceUri string, builder *fileparser.ParserBuilder) (*warningTable, error) {
	tblInfo, err := getTableInfo(ctx, db, schema, table)
	if err != nil {
		return nil, err
	}

	fileParser, err := builder.Build(sourceUri)
	if err != nil {
		plog.L().Warn("create source data parser error", zap.Error(err))
		return nil, nil
	}
	defer fileParser.Close() // nolint: errcheck

	if err := fileParser.ReadRow(); err != nil {
		return nil, errors.Annotate(err, "failed to read row in parser")
	}

	plog.L().Info("check column compatitable", zap.Reflect("file column", fileParser.Columns()),
		zap.Reflect("file last row", fileParser.LastRow()), zap.Reflect("table info", tblInfo), zap.String("sourceUri", sourceUri))
	if err := checkFileColsInDb(tblInfo, fileParser, sourceUri); err != nil {
		return nil, err
	}

	noExistCols, err := findNoExistCols(tblInfo.Cols(), fileParser)
	if err != nil {
		return nil, err
	}

	var warnTable *warningTable
	for _, col := range tblInfo.Cols() {
		if _, ok := noExistCols[col.Name.L]; ok && !columnCanBeEmpty(col) {
			if warnTable == nil {
				warnTable = &warningTable{
					schema: schema,
					table:  table,
					cols:   []string{},
				}
			}

			warnTable.cols = append(warnTable.cols, importcfg.EscapeName(col.Name.L))
		}
	}

	return warnTable, nil
}

func findNoExistCols(colsFromDb []*model.ColumnInfo, fileParser mydump.Parser) (map[string]struct{}, error) {
	colNotExistInFile := make(map[string]struct{}, 0)

	colsFromFile := fileParser.Columns()
	colLenFromFile := len(colsFromFile)
	if colLenFromFile != 0 {
		// check column not exist when column exist in file
		colMapFromFile := make(map[string]struct{}, len(colsFromDb))
		for _, col := range colsFromFile {
			colMapFromFile[col] = struct{}{}
		}
		for _, col := range colsFromDb {
			if _, ok := colMapFromFile[col.Name.L]; !ok {
				colNotExistInFile[col.Name.L] = struct{}{}
			}
		}
	} else {
		// check column not exist when column not exist in file
		colLenFromFile = len(fileParser.LastRow().Row)

		colArray := make([]string, 0, len(colsFromDb))
		for _, col := range colsFromDb {
			colArray = append(colArray, col.Name.L)
		}

		for i := colLenFromFile; i < len(colArray); i++ {
			colNotExistInFile[colArray[i]] = struct{}{}
		}
	}

	return colNotExistInFile, nil
}

func columnCanBeEmpty(col *model.ColumnInfo) bool {
	if pmysql.HasAutoIncrementFlag(col.GetFlag()) || col.IsGenerated() || col.GetDefaultValue() != nil {
		return true
	}
	return !pmysql.HasNotNullFlag(col.GetFlag())
}

func checkFileColsInDb(tableInfo *model.TableInfo, fileParser mydump.Parser, sourceUri string) error {
	colsFromDb := tableInfo.Cols()
	colsFromFile := fileParser.Columns()
	if len(colsFromFile) == 0 {
		colLenFromFile := len(fileParser.LastRow().Row)
		if colLenFromFile > len(colsFromDb) {
			return status.Error(codes.PermissionDenied, fmt.Sprintf("The column count in file %s(%d) exceeds the table %s(%d). Please provide valid files and retry again.", sourceUri, colLenFromFile, tableInfo.Name.L, len(colsFromDb)))
		}
	} else {
		colMapFromDb := make(map[string]struct{}, len(colsFromDb))
		for _, col := range colsFromDb {
			colMapFromDb[col.Name.L] = struct{}{}
		}

		colNotExistInDb := make([]string, 0)
		for _, col := range colsFromFile {
			if _, ok := colMapFromDb[col]; !ok {
				colNotExistInDb = append(colNotExistInDb, importcfg.EscapeName(col))
			}
		}

		if len(colNotExistInDb) > 0 {
			return status.Error(codes.PermissionDenied, fmt.Sprintf("The column(s) %s in file %s does not exist in table %s. Please check CSV file header line and retry it. If the CSV file has a header line, ensure column names of the header line match the target table. If the file doesn't have a header line, disable \"With Header\" in \"CSV configuration\".", strings.Join(colNotExistInDb, ","), importcfg.EscapeName(sourceUri), importcfg.EscapeName(tableInfo.Name.L)))
		}
	}

	return nil
}