package precheck

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

	"github.com/go-sql-driver/mysql"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

const (
	ErrNoSuchTable = 1146
)

type SchemaCheckResult struct {
	Exists bool
	Empty  bool
}

func (res *SchemaCheckResult) IsNotEmpty() bool {
	return res.Exists && !res.Empty
}

func (res *SchemaCheckResult) IsEmpty() bool {
	return res.Exists && res.Empty
}

type SchemaChecker interface {
	CheckSchema(ctx context.Context, database, table string) (*SchemaCheckResult, bizerr.BizError)
}

type DBSchemaChecker struct {
	db *sql.DB
}

func NewDBSchemaChecker(db *sql.DB) *DBSchemaChecker {
	return &DBSchemaChecker{db: db}
}

func (c *DBSchemaChecker) CheckSchema(ctx context.Context, database, table string) (*SchemaCheckResult, bizerr.BizError) {
	sqlQuery := fmt.Sprintf("SELECT 1 FROM `%s`.`%s` USE INDEX() LIMIT 1", EscapeBackticks(database), EscapeBackticks(table))

	var one int // Variable to store the result of SELECT 1
	err := c.db.QueryRowContext(ctx, sqlQuery).Scan(&one)

	if err != nil {
		if isTimeoutError(err) {
			return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
		}
		if errors.Is(err, sql.ErrNoRows) {
			// If QueryRowContext returns ErrNoRows, it means the query executed successfully
			// but found no rows. This indicates the table exists but is empty.
			return &SchemaCheckResult{
				Exists: true,
				Empty:  true,
			}, nil
		}
		var mysqlErr *mysql.MySQLError
		if errors.As(err, &mysqlErr) {
			if mysqlErr.Number == ErrNoSuchTable {
				return &SchemaCheckResult{
					Exists: false,
					Empty:  false,
				}, nil
			}
		}
		return nil, bizerr.WrapError(err, "unknown db query error")
	}

	// If we reach here, it means QueryRowContext successfully scanned a row,
	// implying the table exists and is not empty.
	return &SchemaCheckResult{
		Exists: true,
		Empty:  false,
	}, nil
}

type ClusterProxySchemaChecker struct {
	proxy    clusterproxy.ClusterProxy
	operator *clusterproxy.ClusterOperator
}

func NewClusterProxySchemaChecker(
	proxy clusterproxy.ClusterProxy,
	operator *clusterproxy.ClusterOperator,
) *ClusterProxySchemaChecker {
	return &ClusterProxySchemaChecker{proxy: proxy, operator: operator}
}

// CheckSchema checks if a table exists and is empty using the provided proxy.
// It returns a SchemaCheckResult and any error that occurred during the check.
func (c *ClusterProxySchemaChecker) CheckSchema(ctx context.Context, database, table string) (*SchemaCheckResult, bizerr.BizError) {
	// Here we use the `USE INDEX()` hint to skip fetch the record from index.
	// In Lightning, if previous importing is halted half-way, it is possible that
	// the data is partially imported, but the index data has not been imported.
	// In this situation, if no hint is added, the SQL executor might fetch the record from index,
	// which is empty.  This will result in missing check.
	sql := fmt.Sprintf("SELECT 1 FROM `%s`.`%s` USE INDEX() LIMIT 1", EscapeBackticks(database),
		EscapeBackticks(table))
	res, err := c.proxy.QuerySQLWithSqlError(ctx, c.operator, sql)
	if err != nil {
		if isTimeoutError(err) {
			return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
		}
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s", err.Error())
	}
	if res == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "query result is nil")
	}
	if res.SqlError.ErrNumber == ErrNoSuchTable {
		return &SchemaCheckResult{
			Exists: false,
			Empty:  false,
		}, nil
	}

	// If table is not empty, will return {\"1\":[1]}, otherwise will return {}
	if !strings.EqualFold(string(res.Result), "{}") {
		return &SchemaCheckResult{
			Exists: true,
			Empty:  false,
		}, nil
	}
	return &SchemaCheckResult{
		Exists: true,
		Empty:  true,
	}, nil
}

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