package main

import (
	"context"
	"database/sql"
	"flag"
	"fmt"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/juju/errors"
	"github.com/pingcap/schrodinger-test/util"
	client "github.com/pingcap/schrodinger/pkg/test_client"
	"k8s.io/apimachinery/pkg/util/wait"
)

var (
	gcTime        time.Duration
	checkInterval time.Duration
)

func init() {
	flag.DurationVar(&gcTime, "gc-time", 720*time.Hour, "set tidb cluster gc time")
	flag.DurationVar(&checkInterval, "check-interval", 2*time.Minute, "the interval to check table")

	flag.Parse()
}

func main() {
	client.Info("[admin_check] start ...")

	ctx, cancel := context.WithCancel(context.Background())
	cli := client.New(ctx, client.ModeOnline)

	dbAddr := cli.TiDBService()
	dbDSN := fmt.Sprintf("root:@tcp(%s)/test", dbAddr)
	db, err := util.OpenDB(dbDSN, 1)
	if err != nil {
		client.Fatalf("[admin_check] create db client error %v", err)
	}

	sc := make(chan os.Signal, 1)
	signal.Notify(sc,
		syscall.SIGHUP,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT)
	go func() {
		sig := <-sc
		client.Infof("[admin_check] Got signal [%s] to exist.", sig)
		cancel()
		os.Exit(0)
	}()

	if err := adjustTiDBGC(ctx, db); err != nil {
		client.Fatalf("[admin_check] return with error: %v", err)
	}

	if err := runCheck(ctx, db); err != nil {
		client.Fatalf("[admin_check] return with error %v", err)
	}
}

func adjustTiDBGC(ctx context.Context, db *sql.DB) error {
	err := wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		stmt := fmt.Sprintf("update mysql.tidb set variable_value='%s' where variable_name='tikv_gc_life_time'; "+
			"update mysql.tidb set variable_value='%s' where variable_name='tikv_gc_run_interval';", gcTime.String(), gcTime.String())
		_, err := db.Exec(stmt)
		if err != nil {
			client.Warnf("[admin_check] adjusttidbgc:%s failed: %v, retry", stmt, err)
			return false, nil
		}
		return true, nil
	})

	if err != nil {
		client.Errorf("[admin_check] adjuet tidb GC failed: %v", err)
		return errors.Trace(err)
	}

	return nil
}

func runCheck(ctx context.Context, db *sql.DB) error {
	ticker := time.NewTicker(checkInterval)
	for {
		select {
		case <-ctx.Done():
			client.Info("[admin_check] is canceled, exit")
			return nil
		case <-ticker.C:
			if err := checkAction(db); err != nil {
				client.Errorf("[admin_check] check failed: %v, waiting for next", err)
			}
		}
	}
}

func checkAction(db *sql.DB) error {
	databases, err := getAllDatabases(db)
	if err != nil {
		client.Errorf("[admin_check] get databases failed: %v", err)
		return errors.Trace(err)
	}

	for _, dbName := range databases {
		if err = checkDatabase(db, dbName); err != nil {
			client.Errorf("[admin_check] check database %s failed: %v", dbName, err)
			continue
		}
	}

	return nil
}

func checkDatabase(db *sql.DB, dbName string) error {
	useStmt := fmt.Sprintf("USE %s", dbName)
	if _, err := db.Exec(useStmt); err != nil {
		client.Errorf("[admin_check] exec %s failed: %v", useStmt, err)
		return errors.Trace(err)
	}

	err := addTable(db)
	if err != nil {
		client.Errorf("add table failed %v", err)
		return errors.Trace(err)
	}

	tables, err := getAllTables(db)
	if err != nil {
		client.Errorf("[admin_check] databases %s get tables failed: %v", dbName, err)
		return errors.Trace(err)
	}

	for _, table := range tables {
		client.Infof("[admin_check] start to check %s.%s", dbName, table)
		if err = checkTable(db, dbName, table); err != nil {
			client.Errorf("[admin_check] check %s.%s failed: %v", dbName, table, err)
			continue
		}
	}

	return nil
}

func checkTable(db *sql.DB, dbName, table string) error {
	stmt := fmt.Sprintf("admin check table `%s`", table)
	client.Infof("[admin_check] start to exec %s", stmt)
	_, err := db.Exec(stmt)
	if err != nil && strings.Contains(err.Error(), "1105") && !getFilter(err) {
		client.Fatalf("[admin_check] check %s.%s data nonconsistent %v", dbName, table, err)
	}

	return errors.Trace(err)
}

func getAllDatabases(db *sql.DB) ([]string, error) {
	databases := []string{}
	stmt := "SHOW DATABASES"
	rows, err := db.Query(stmt)
	if err != nil {
		client.Errorf("[admin_check] exec %s failed: %v", stmt, err)
		return databases, errors.Trace(err)
	}

	for rows.Next() {
		var dbName string
		if err := rows.Scan(&dbName); err != nil {
			client.Errorf("[admin_check] scan dbname failed: %v", err)
			return databases, errors.Trace(err)
		}
		databases = append(databases, dbName)
	}

	return databases, nil
}

func getAllTables(db *sql.DB) ([]string, error) {
	tables := []string{}
	stmt := "SHOW TABLES"
	rows, err := db.Query(stmt)
	if err != nil {
		client.Errorf("[admin_check] exec %s failed: %v", stmt, err)
		return tables, errors.Trace(err)
	}

	for rows.Next() {
		var table string
		if err := rows.Scan(&table); err != nil {
			client.Errorf("[admin_check] scan table failed: %v", err)
			return tables, errors.Trace(err)
		}
		tables = append(tables, table)
	}

	return tables, nil
}

func addTable(db *sql.DB) error {
	stmt := "drop table if exists check_table_t1;CREATE TABLE check_table_t1 (c2 BOOL, PRIMARY KEY (c2));INSERT INTO check_table_t1 SET c2 = '0';ALTER TABLE check_table_t1 ADD COLUMN c3 DATETIME NULL DEFAULT '2668-02-03 17:19:31';ALTER TABLE check_table_t1 ADD INDEX idx2 (c3);"
	client.Infof("[admin_check] start to add table %s", stmt)
	_, err := db.Exec(stmt)
	if err != nil && strings.Contains(err.Error(), "1105") && !getFilter(err) {
		client.Fatalf("[admin_check] add table check_table_t1 data nonconsistent %v", err)
	}

	return errors.Trace(err)
}

func getFilter(err error) bool {
	return strings.Contains(err.Error(), "cancelled DDL job") || strings.Contains(err.Error(), "Information schema is changed") ||
		strings.Contains(err.Error(), "TiKV server timeout") || strings.Contains(err.Error(), "Error 9007: Write conflict")
}