package consistency

import (
	"database/sql"
	"fmt"
	"net/url"

	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/mysql"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

type Checker struct {
	upStream   *url.URL
	downStream *url.URL
	dbName     string
	snapshotTs uint64
}

// NewChecker return a consistency Checker, `defer ShouldClose()` should be called immediately.
// source and sink db instance should specify the database name.
func NewChecker(source, sink *url.URL, dbName string, snapshotTs uint64) *Checker {
	return &Checker{
		upStream:   source,
		downStream: sink,
		dbName:     dbName,
		snapshotTs: snapshotTs,
	}
}

// Check the consistency between upstream and downstream db
func (c *Checker) Check() error {
	sourceDB, err := sql.Open("mysql", resource.BuildDSN(c.upStream, c.dbName))
	if err != nil {
		return err
	}
	if c.snapshotTs > 0 {
		query := fmt.Sprintf(`set @@tidb_snapshot=%d`, c.snapshotTs)
		_, err = sourceDB.Exec(query)
		if err != nil {
			return err
		}
		logger.Info("check consistency between upstream and downstream db", zap.String("query", query))
	}

	sinkDB, err := sql.Open("mysql", resource.BuildDSN(c.downStream, c.dbName))
	if err != nil {
		return err
	}
	defer func() {
		if err := sourceDB.Close(); err != nil {
			logger.Error("close upstream db failed", zap.Error(err))
		}
		if err := sinkDB.Close(); err != nil {
			logger.Error("close downstream db failed", zap.Error(err))
		}
	}()
	// check consistency by crc32
	return CompareCRC32CheckSum(sourceDB, sinkDB)
}

func (c *Checker) String() string {
	return fmt.Sprintf("upstream=%+v, downstream=%+v", c.upStream, c.downStream)
}

type ConsistencyChecker interface {
	CheckTable(database, table string, expected interface{}) bool
	CheckDatabase(database string, expected interface{}) bool
	CheckDatabases(databases []string, expected interface{}) bool
	CheckUserDatabases(expected interface{}) bool
}

type Crc32Checker struct {
	*mysql.MySQLClient
}

func (c *Crc32Checker) CheckTable(database, table string, expected interface{}) bool {
	got := utils.Try(c.Crc32Table(database, table)).(uint32)
	want := expected.(uint32)
	if got != want {
		log.Error("table crc32 checksum is not matched", zap.String("database", database), zap.String("table", table), zap.Uint32("want", want), zap.Uint32("got", got))
		return false
	}
	return true
}

func (c *Crc32Checker) CheckDatabase(database string, expected interface{}) bool {
	got := utils.Try(c.Crc32Database(database, 10)).(uint32)
	want := expected.(uint32)
	if got != want {
		log.Error("database crc32 checksum is not matched", zap.String("database", database), zap.Uint32("want", want), zap.Uint32("got", got))
		return false
	}
	return true
}

func (c *Crc32Checker) CheckDatabases(databases []string, expected interface{}) bool {
	got := utils.Try(c.Crc32Databases(databases, 10)).(uint32)
	want := expected.(uint32)
	if got != want {
		log.Error("databases crc32 checksum is not matched", zap.Strings("databases", databases), zap.Uint32("want", want), zap.Uint32("got", got))
		return false
	}
	return true
}

func (c *Crc32Checker) CheckUserDatabases(expected interface{}) bool {
	got := utils.Try(c.Crc32UserDatabases(10)).(uint32)
	want := expected.(uint32)
	log.Info("compare crc32", zap.Uint32("want", want), zap.Uint32("got", got))
	if got != want {
		log.Error("user databases crc32 checksum is not matched", zap.Uint32("want", want), zap.Uint32("got", got))
		return false
	}
	return true
}

func (c *Crc32Checker) CheckUserDatabasesConsistency(otherClient *mysql.MySQLClient) bool {
	got := utils.Try(c.Crc32UserDatabases(10)).(uint32)
	want := utils.Try(otherClient.Crc32UserDatabases(10)).(uint32)
	log.Info("compare crc32", zap.Uint32("want", want), zap.Uint32("got", got))
	if got != want {
		log.Error("user databases crc32 checksum is not matched", zap.Uint32("want", want), zap.Uint32("got", got))
		return false
	}
	return true
}

var _ ConsistencyChecker = &Crc32Checker{}

type binlogChecker struct {
	upStream   *url.URL
	downStream *url.URL
	dbName     string
}

// Use for Checker without snapshot TS
// source and sink db instance should specify the database name.
func NewBinlogChecker(source, sink *url.URL, dbName string) *binlogChecker {
	return &binlogChecker{
		upStream:   source,
		downStream: sink,
		dbName:     dbName,
	}
}

// Check the consistency between upstream and downstream db
func (c *binlogChecker) Check() error {
	sourceDB, err := sql.Open("mysql", resource.BuildDSN(c.upStream, c.dbName))
	if err != nil {
		return err
	}

	sinkDB, err := sql.Open("mysql", resource.BuildDSN(c.downStream, c.dbName))
	if err != nil {
		return err
	}
	defer func() {
		if err := sourceDB.Close(); err != nil {
			logger.Error("close upstream db failed", zap.Error(err))
		}
		if err := sinkDB.Close(); err != nil {
			logger.Error("close downstream db failed", zap.Error(err))
		}
	}()
	// check consistency by crc32
	return CompareCRC32CheckSum(sourceDB, sinkDB)
}

func (c *binlogChecker) String() string {
	return fmt.Sprintf("upstream=%+v, downstream=%+v", c.upStream, c.downStream)
}