package steps

import (
	"context"
	"database/sql"
	"fmt"
	"net/url"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/consistency"
	"github.com/pingcap/test-infra/caselib/pkg/model/ticdc"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"github.com/pingcap/test-infra/caselib/pkg/mysql"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/pingcap/tiflow/pkg/retry"
	"go.uber.org/zap"
)

type sendFinishMark struct {
	dbName string
}

func NewSendFinishMark(dbName string) Step {
	return &sendFinishMark{
		dbName: dbName,
	}
}

func (s *sendFinishMark) Execute(ctx context.Context) error {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return errors.Trace(err)
	}

	db, err := tc.GetDB(s.dbName)
	if err != nil {
		log.Error("send finish mark failed", zap.Error(err))
		return errors.Trace(err)
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	// send a DDL, this can be used for project TiCDC, which use DDL as a strong sync point
	query := "CREATE TABLE IF NOT EXISTS finishmark (foo BIGINT PRIMARY KEY)"
	execF := func() error {
		_, err := db.ExecContext(context.Background(), query)
		return errors.Trace(err)
	}

	if err := retry.Do(ctx, execF, retry.WithBackoffBaseDelay(3000), retry.WithMaxTries(10)); err != nil {
		log.Error("send finish mark failed", zap.Error(err), zap.String("dbName", s.dbName))
		return errors.Trace(err)
	}

	return nil
}

func (s *sendFinishMark) String() string {
	return fmt.Sprintf("sendFinishMark: dbName=%+v", s.dbName)
}

type sendFinishMarkTiup struct {
	dbName string
}

func NewSendFinishMarkTiup(dbName string) Step {
	return &sendFinishMarkTiup{
		dbName: dbName,
	}
}

func (s *sendFinishMarkTiup) Execute(ctx context.Context) error {
	db, err := tiupTiDBCluster.GetDB(s.dbName)
	if err != nil {
		log.Error("send finish mark to tiup cluster failed", zap.Error(err))
		return errors.Trace(err)
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db for tiup cluster failed", zap.Error(err))
		}
	}()

	// send a DDL, this can be used for project TiCDC, which use DDL as a strong sync point
	query := "CREATE TABLE IF NOT EXISTS finishmark (foo BIGINT PRIMARY KEY)"
	execF := func() error {
		_, err := db.ExecContext(context.Background(), query)
		return errors.Trace(err)
	}

	if err := retry.Do(ctx, execF, retry.WithBackoffBaseDelay(3000), retry.WithMaxTries(10)); err != nil {
		log.Error("send finish mark to tiup cluster failed", zap.Error(err), zap.String("dbName", s.dbName))
		return errors.Trace(err)
	}

	return nil
}

func (s *sendFinishMarkTiup) String() string {
	return fmt.Sprintf("sendFinishMarkTiup: dbName=%+v", s.dbName)
}

type waitFinishMark struct {
	handle  *apiresource.ResourceHandle
	dbName  string
	timeout time.Duration
}

func (w *waitFinishMark) String() string {
	return fmt.Sprintf("waitFinishMark: db=%+v, dbName=%+v, timeout=%+v", w.handle, w.dbName, w.timeout)
}

func (w *waitFinishMark) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(w.handle)
	if err != nil {
		return errors.Trace(err)
	}
	mysql, ok := resolved.(resource.MySQLCompatible)
	if !ok {
		return fmt.Errorf("wait finish mark should on a MySQL Compatible(TiDB / MySQL), but got: %+v", resolved)
	}
	db, err := mysql.GetDB(w.dbName)
	if err != nil {
		return errors.Trace(err)
	}

	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	var (
		start        = time.Now()
		ticker       = time.NewTicker(10 * time.Second)
		timeoutTimer = time.NewTimer(w.timeout)
	)
	defer func() {
		ticker.Stop()
		timeoutTimer.Stop()
	}()

	log.Info("wait for finishmark", zap.Time("start", start), zap.Duration("timeout", w.timeout))
loop:
	for {
		select {
		case <-timeoutTimer.C:
			log.Warn("wait for finishmark failed, timeout")
			return fmt.Errorf("failed to wait for finish mark")
		case <-ticker.C:
			if ok := isTableExist(context.Background(), db); ok {
				log.Info("finishmark synced to down stream", zap.Duration("elapsed", time.Since(start)))
				break loop
			}
		}
	}

	return nil
}

func isTableExist(ctx context.Context, db *sql.DB) bool {
	query := "SHOW TABLES LIKE 'finishmark'"
	var t string
	err := db.QueryRowContext(ctx, query).Scan(&t)
	if err == sql.ErrNoRows {
		return false
	}

	// we don't know what kind of error would happen if we inject chaos into TiDB Cluster
	// so we would prefer to retry and log the error
	// finish mark may already there, but we do not return since meet other chaos caused error.
	if err != nil {
		log.Warn("query finishmark at sink db failed", zap.Error(err))
		return false
	}
	return true
}

// NewWaitFinishMark wait for finish mark send to the target db.
func NewWaitFinishMark(handle *apiresource.ResourceHandle, dbName string, timeout time.Duration) Step {
	return &waitFinishMark{
		handle:  handle,
		dbName:  dbName,
		timeout: timeout,
	}
}

type checkConsistency struct {
	source *apiresource.ResourceHandle
	target *apiresource.ResourceHandle
	dbName string
}

func NewCheckConsistency(source, target *apiresource.ResourceHandle, dbName string) Step {
	return &checkConsistency{
		source: source,
		target: target,
		dbName: dbName,
	}
}

func (c *checkConsistency) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(c.source)
	if err != nil {
		return errors.Trace(err)
	}
	tc, ok := resolved.(resource.MySQLCompatible)
	if !ok {
		return fmt.Errorf("consistency check source should be MySQL Compatible(TiDB), but got: %+v", resolved)
	}
	upstreamURL, err := tc.ServiceURL(resource.DBAddr)
	if err != nil {
		return errors.Trace(err)
	}

	resolved, err = core.ResolveResourceHandle(c.target)
	if err != nil {
		return errors.Trace(err)
	}
	mysql, ok := resolved.(resource.MySQLCompatible)
	if !ok {
		return fmt.Errorf("consistency check target should be MySQL Compatible(MySQL / TiDB), but got: %+v", resolved)
	}
	downstreamURL, err := mysql.ServiceURL(resource.DBAddr)
	if err != nil {
		return errors.Trace(err)
	}

	var snapshotTs uint64
	meta, ok := GetInner(ctx).Board().value.(*ticdc.RedoMeta)
	if ok {
		snapshotTs = meta.ResolvedTs
	}

	return consistency.NewChecker(upstreamURL, downstreamURL, c.dbName, snapshotTs).Check()
}

func (c *checkConsistency) String() string {
	return fmt.Sprintf("checkConsistency: dbName=%+v", c.dbName)
}

type checkConsistencyTiup struct {
	tiupClusterOption workload.TiupClusterOption
	dbName            string
}

func NewCheckConsistencyTiup(tiupClusterOption workload.TiupClusterOption, dbName string) Step {
	return &checkConsistencyTiup{
		tiupClusterOption: tiupClusterOption,
		dbName:            dbName,
	}
}

func (c *checkConsistencyTiup) Execute(ctx context.Context) error {
	sink, err := GetInner(ctx).Board().Sink()
	if err != nil {
		return err
	}

	upstreamURL, err := tiupTiDBCluster.ServiceURL(resource.DBAddr)
	if err != nil {
		return err
	}

	// todo: check consistency only works for MySQL now.
	// add a consistency check method for kafka.
	downstreamURL, err := sink.SinkEndpoint()
	if err != nil {
		return err
	}
	return consistency.NewBinlogChecker(upstreamURL, downstreamURL, c.dbName).Check()
}

func (c *checkConsistencyTiup) String() string {
	return fmt.Sprintf("checkConsistencyTiup: dbName=%+v", c.dbName)
}

type checkConsistencyBinlogTiup struct {
	tiupClusterOption workload.TiupClusterOption
	downStream        *apiresource.ResourceHandle
	dbName            string
}

func NewCheckConsistencyBinlogTiup(tiupClusterOption workload.TiupClusterOption, downStream *apiresource.ResourceHandle, dbName string) Step {
	return &checkConsistencyBinlogTiup{
		tiupClusterOption: tiupClusterOption,
		downStream:        downStream,
		dbName:            dbName,
	}
}

func (c *checkConsistencyBinlogTiup) Execute(ctx context.Context) error {
	down := utils.Try(core.ResolveResourceHandle(c.downStream)).(resource.MySQLCompatible)
	downstreamURL := utils.Try(down.ServiceURL(resource.DBAddr)).(*url.URL)

	upstreamURL, err := tiupTiDBCluster.ServiceURL(resource.DBAddr)
	if err != nil {
		return err
	}

	// todo: check consistency only works for MySQL now.
	// add a consistency check method for kafka.
	// downstreamURL, err := sink.SinkEndpoint()
	// if err != nil {
	// 	return err
	// }
	return consistency.NewBinlogChecker(upstreamURL, downstreamURL, c.dbName).Check()
}

func (c *checkConsistencyBinlogTiup) String() string {
	return fmt.Sprintf("checkConsistencyBinlogTiup: dbName=%+v", c.dbName)
}

type checkUserDatabasesCrc32 struct {
	mysqlRes *apiresource.ResourceHandle
	expected *uint32
}

func (c checkUserDatabasesCrc32) String() string {
	if *c.expected == 0 {
		return fmt.Sprintf("Skip checking crc32")
	}
	return fmt.Sprintf("Check user databases crc32, expected: %d", *c.expected)
}

func (c checkUserDatabasesCrc32) Execute(ctx context.Context) error {
	if *c.expected == 0 {
		return nil
	}
	mysqlRes := utils.Try(core.ResolveResourceHandle(c.mysqlRes)).(resource.MySQLCompatible)
	db := utils.Try(mysqlRes.GetDB("")).(*sql.DB)
	crc32Checker := consistency.Crc32Checker{MySQLClient: mysql.NewMySQLClient(db)}
	if !crc32Checker.CheckUserDatabases(*c.expected) {
		return errors.New("checksum mismatched")
	}
	return nil
}

var _ Step = &checkUserDatabasesCrc32{}

func (b *Builder) CheckUserDatabasesCrc32(expected *uint32, mysqlRes *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkUserDatabasesCrc32{mysqlRes: mysqlRes, expected: expected})
	return b
}

type checkDatabasesCrc32 struct {
	mysqlRes  *apiresource.ResourceHandle
	expected  *uint32
	databases *string
}

func (c checkDatabasesCrc32) String() string {
	if *c.expected == 0 {
		return fmt.Sprintf("Skip checking crc32")
	}
	return fmt.Sprintf("Check databases crc32, databases: %s, expected: %d", *c.databases, *c.expected)
}

func (c checkDatabasesCrc32) Execute(ctx context.Context) error {
	if *c.expected == 0 {
		return nil
	}
	mysqlRes := utils.Try(core.ResolveResourceHandle(c.mysqlRes)).(resource.MySQLCompatible)
	db := utils.Try(mysqlRes.GetDB("")).(*sql.DB)
	databases := strings.Split(*c.databases, ",")
	crc32Checker := consistency.Crc32Checker{MySQLClient: mysql.NewMySQLClient(db)}
	if !crc32Checker.CheckDatabases(databases, *c.expected) {
		return errors.New("checksum mismatched")
	}
	return nil
}

var _ Step = &checkDatabasesCrc32{}

func (b *Builder) CheckDatabasesCrc32(expected *uint32, databases *string, mysqlRes *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkDatabasesCrc32{mysqlRes, expected, databases})
	return b
}

type checkTableCrc32 struct {
	mysqlRes *apiresource.ResourceHandle
	expected *uint32
	database *string
	table    *string
}

func (c checkTableCrc32) String() string {
	if *c.expected == 0 {
		return fmt.Sprintf("Skip checking crc32")
	}
	return fmt.Sprintf("Check table crc32, table: %s, expected: %d", *c.table, *c.expected)
}

func (c checkTableCrc32) Execute(ctx context.Context) error {
	if *c.expected == 0 {
		return nil
	}
	mysqlRes := utils.Try(core.ResolveResourceHandle(c.mysqlRes)).(resource.MySQLCompatible)
	db := utils.Try(mysqlRes.GetDB("")).(*sql.DB)
	crc32Checker := consistency.Crc32Checker{MySQLClient: mysql.NewMySQLClient(db)}
	if !crc32Checker.CheckTable(*c.database, *c.table, *c.expected) {
		return errors.New("checksum mismatched")
	}
	return nil
}

var _ Step = &checkTableCrc32{}

func (b *Builder) CheckTableCrc32(expected *uint32, database, table *string, mysqlRes *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkTableCrc32{mysqlRes, expected, database, table})
	return b
}

type checkUserDatabasesConsistency struct {
	upstream   *apiresource.ResourceHandle
	downstream *apiresource.ResourceHandle
}

func (c *checkUserDatabasesConsistency) Execute(ctx context.Context) error {
	mysqlRes := utils.Try(core.ResolveResourceHandle(c.upstream)).(resource.MySQLCompatible)
	db := utils.Try(mysqlRes.GetDB("")).(*sql.DB)
	otherRes := utils.Try(core.ResolveResourceHandle(c.downstream)).(resource.MySQLCompatible)
	otherDb := utils.Try(otherRes.GetDB("")).(*sql.DB)
	crc32Checker := consistency.Crc32Checker{MySQLClient: mysql.NewMySQLClient(db)}
	otherClient := mysql.NewMySQLClient(otherDb)
	if !crc32Checker.CheckUserDatabasesConsistency(otherClient) {
		return errors.New("checksum mismatched")
	}
	return nil
}

func (c *checkUserDatabasesConsistency) String() string {
	return fmt.Sprintf("checkUserDatabasesConsistency, upstream: %+v, downstream: %+v", c.upstream, c.downstream)
}

func NewCheckUserDatabasesConsistency(upstream, downstream *apiresource.ResourceHandle) Step {
	return &checkUserDatabasesConsistency{upstream, downstream}
}

type sendBinlogFinishMark struct {
	tidb   *apiresource.ResourceHandle
	dbName string
}

func NewSendBinlogFinishMark(tidbcluster *apiresource.ResourceHandle, dbName string) Step {
	return &sendBinlogFinishMark{
		tidb:   tidbcluster,
		dbName: dbName,
	}
}

func (s *sendBinlogFinishMark) Execute(ctx context.Context) error {
	up := utils.Try(core.ResolveResourceHandle(s.tidb)).(resource.MySQLCompatible)
	db := utils.Try(up.GetDB(s.dbName)).(*sql.DB)
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	// send a DDL, this can be used for project Binlog, which use DDL as a strong sync point
	query := "CREATE TABLE IF NOT EXISTS finishmark (foo BIGINT PRIMARY KEY)"
	execF := func() error {
		_, err := db.ExecContext(context.Background(), query)
		return errors.Trace(err)
	}

	if err := retry.Do(ctx, execF, retry.WithBackoffBaseDelay(3000), retry.WithMaxTries(10)); err != nil {
		log.Error("send finish mark failed", zap.Error(err), zap.String("dbName", s.dbName))
		return errors.Trace(err)
	}

	return nil
}

func (s *sendBinlogFinishMark) String() string {
	return fmt.Sprintf("sendBinlogFinishMark: dbName=%+v", s.dbName)
}

func (b *Builder) SendBinlogFinishMark(tidb *apiresource.ResourceHandle, dbName string) *Builder {
	b.steps = append(b.steps, NewSendBinlogFinishMark(tidb, dbName))
	return b
}

type waitBinlogFinishMark struct {
	tidb    *apiresource.ResourceHandle
	dbName  string
	timeout time.Duration
}

func (w *waitBinlogFinishMark) String() string {
	return fmt.Sprintf("binlogWaitFinishMark: dbName=%+v, timeout=%+v", w.dbName, w.timeout)
}

func (w *waitBinlogFinishMark) Execute(ctx context.Context) error {
	downStream := utils.Try(core.ResolveResourceHandle(w.tidb)).(resource.MySQLCompatible)
	dbURL := utils.Try(downStream.ServiceURL(resource.DBAddr)).(*url.URL)

	// todo: after kafka supported, at a new sink interface,
	//  also replace a db to check for downstream finish mark and consistency
	//  sink would be mysql / tidb / kafka, but we should check finish table only for db
	db, err := sql.Open("mysql", resource.BuildDSN(dbURL, w.dbName))
	if err != nil {
		return errors.Trace(err)
	}
	if err := db.Ping(); err != nil {
		return errors.Trace(err)
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	var (
		start        = time.Now()
		ticker       = time.NewTicker(30 * time.Second)
		timeoutTimer = time.NewTimer(w.timeout)
	)
	defer func() {
		ticker.Stop()
		timeoutTimer.Stop()
	}()

	log.Info("wait for finishmark", zap.Time("start", start), zap.Duration("timeout", w.timeout))
loop:
	for {
		select {
		case <-timeoutTimer.C:
			log.Warn("wait for finishmark failed, timeout")
			return fmt.Errorf("failed to wait for finish mark")
		case <-ticker.C:
			if ok := isTableExist(context.Background(), db); ok {
				log.Info("finishmark synced to down stream", zap.Duration("elapsed", time.Since(start)))
				break loop
			}
		}
	}

	return nil
}

// NewWaitFinishMark wait for finish mark send to the target db, if haveNemesis, stop nemesis if failed.
func NewWaitBinlogFinishMark(tidbcluster *apiresource.ResourceHandle, dbName string, timeout time.Duration) Step {
	return &waitBinlogFinishMark{
		tidb:    tidbcluster,
		dbName:  dbName,
		timeout: timeout,
	}
}

func (b *Builder) WaitBinlogFinishMark(tidb *apiresource.ResourceHandle, dbName string, timeout time.Duration) *Builder {
	b.steps = append(b.steps, NewWaitBinlogFinishMark(tidb, dbName, timeout))
	return b
}

type checkBinlogConsistency struct {
	upStream   *apiresource.ResourceHandle
	downStream *apiresource.ResourceHandle
	dbName     string
}

func NewCheckBinlogConsistency(upStream, downStream *apiresource.ResourceHandle, dbName string) Step {
	return &checkBinlogConsistency{
		upStream:   upStream,
		downStream: downStream,
		dbName:     dbName,
	}
}

func (c *checkBinlogConsistency) Execute(ctx context.Context) error {
	down := utils.Try(core.ResolveResourceHandle(c.downStream)).(resource.MySQLCompatible)
	downURL := utils.Try(down.ServiceURL(resource.DBAddr)).(*url.URL)

	up := utils.Try(core.ResolveResourceHandle(c.upStream)).(resource.MySQLCompatible)
	upURL := utils.Try(up.ServiceURL(resource.DBAddr)).(*url.URL)

	return consistency.NewBinlogChecker(upURL, downURL, c.dbName).Check()
}

func (c *checkBinlogConsistency) String() string {
	return fmt.Sprintf("checkConsistency: dbName=%+v", c.dbName)
}

func (b *Builder) CheckBinlogConsistency(upStream, downStream *apiresource.ResourceHandle, dbName string) *Builder {
	b.steps = append(b.steps, NewCheckBinlogConsistency(upStream, downStream, dbName))
	return b
}