package ddl

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

	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

func AddIndexForTable(ctx context.Context, tc resource.MySQLCompatible, timeout time.Duration, dbName string, tableName string, addIndexColumnName string) error {
	timeOut := time.After(timeout)
	errCh := make(chan error)
	go func(errCh chan error) {
		dbUrl := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
		db, err := sql.Open("mysql", resource.BuildDSN(dbUrl, dbName, "timeout=300m", "readTimeout=300m", "writeTimeout=300m"))
		Expect(err).Should(Succeed())
		defer db.Close()
		log.Info("add index", zap.String("tableName", tableName), zap.String("addIndexColumnName", addIndexColumnName))
		var indexName string
		cmd := fmt.Sprintf("select column_name from information_schema.statistics where table_schema='%s' and table_name='%s' and column_name='%s';", dbName, tableName, addIndexColumnName)
		ret := db.QueryRow(cmd).Scan(&indexName)
		log.Info(indexName)
		dropCmd := fmt.Sprintf("alter table %s drop index sys_eidx;", tableName)
		addCmd := fmt.Sprintf("alter table %s add index sys_eidx (%s);", tableName, addIndexColumnName)
		if ret == nil {
			err = oltppkg.DbExecContextWithLog(ctx, db, dropCmd)
			if err != nil && !errors.Is(err, context.DeadlineExceeded) {
				errCh <- err
				return
			}
		}
		err = oltppkg.DbExecContextWithLog(ctx, db, addCmd)
		if err != nil && !errors.Is(err, context.DeadlineExceeded) {
			errCh <- err
			return
		}
		err = oltppkg.DbExecContextWithLog(ctx, db, dropCmd)
		if err != nil && !errors.Is(err, context.DeadlineExceeded) {
			errCh <- err
			return
		}
	}(errCh)
	for {
		select {
		case <-ctx.Done():
			log.Info("add index ctx done.")
			return nil
		case <-timeOut:
			log.Info("add index ctx timeout.")
			return nil
		case err := <-errCh:
			Expect(err).Should(Succeed())
			return err
		}
	}
}

func AddColForAllTables(ctx context.Context, tc resource.MySQLCompatible, db *sql.DB, tables []string, addColumnName string) error {
	tableNum := len(tables)
	log.Info("add colum for all tables")
	for i := 0; i < tableNum; i++ {
		tableName := tables[i]
		addIndex := fmt.Sprintf("alter table %s add column %s int;", tableName, addColumnName)
		Try(DbExecWithIgnoreCtxDeadline(ctx, db, addIndex))
	}
	return nil
}

func DropColForAllTables(ctx context.Context, tc resource.MySQLCompatible, db *sql.DB, tables []string, addColumnName string) error {
	tableNum := len(tables)
	log.Info("drop colum for all tables")
	for i := 0; i < tableNum; i++ {
		tableName := tables[i]
		dropIndex := fmt.Sprintf("alter table %s drop column %s;", tableName, addColumnName)
		Try(DbExecWithIgnoreCtxDeadline(ctx, db, dropIndex))
	}
	return nil
}

func RecoverColForAllTables(ctx context.Context, tc resource.MySQLCompatible, dbName string, addColumnName string) error {
	db := Try(ConnectDbWithTimeout(tc, dbName, "timeout=300m", "readTimeout=300m", "writeTimeout=300m")).(*sql.DB)
	defer db.Close()
	tables := oltppkg.GetAllTablesNames(ctx, tc, dbName)
	tableNum := len(tables)
	log.Info("recover colum for all tables")
	for i := 0; i < tableNum; i++ {
		var columeName string
		tableName := tables[i]
		cmd := fmt.Sprintf("select column_name from information_schema.columns where table_schema='%s' and table_name='%s' and column_name='%s';", dbName, tableName, addColumnName)
		ret := db.QueryRow(cmd).Scan(&columeName)
		dropIndex := fmt.Sprintf("alter table %s drop column %s;", tableName, addColumnName)
		if ret == nil {
			Try(DbExecWithIgnoreCtxDeadline(ctx, db, dropIndex))
		}
	}
	return nil
}

func AddColForAllTablesRepeatedly(ctx context.Context, tc resource.MySQLCompatible, timeout time.Duration, dbName string, addColumnName string) error {
	db := Try(ConnectDbWithTimeout(tc, dbName, "timeout=300m", "readTimeout=300m", "writeTimeout=300m")).(*sql.DB)
	defer db.Close()
	tables := oltppkg.GetAllTablesNames(ctx, tc, dbName)
	timeOut := time.After(timeout)
	log.Info("start add and drop column for all tables Repeatedly")
	for {
		select {
		case <-ctx.Done():
			log.Info("and column ctx done.")
			return nil
		case <-timeOut:
			log.Info("and column ctx timeout.")
			return nil
		default:
		}
		AddColForAllTables(ctx, tc, db, tables, addColumnName)
		DropColForAllTables(ctx, tc, db, tables, addColumnName)
	}
}

func AnalyzeAllTables(ctx context.Context, tc resource.MySQLCompatible, db *sql.DB, tables []string) error {
	tableNum := len(tables)
	log.Info("analyze all tables")
	for i := 0; i < tableNum; i++ {
		tableName := tables[i]
		analyzeTable := fmt.Sprintf("analyze table %s;", tableName)
		Try(DbExecWithIgnoreCtxDeadline(ctx, db, analyzeTable))
	}
	return nil
}

func AnalyzeAllTablesRepeatedly(ctx context.Context, tc resource.MySQLCompatible, timeout time.Duration, dbName string) error {
	db := Try(ConnectDbWithTimeout(tc, dbName, "timeout=300m", "readTimeout=300m", "writeTimeout=300m")).(*sql.DB)
	defer db.Close()
	tables := oltppkg.GetAllTablesNames(ctx, tc, dbName)
	timeOut := time.After(timeout)
	log.Info("start analyze for all tables Repeatedly")
	for {
		select {
		case <-ctx.Done():
			log.Info("analyze ctx done.")
			return nil
		case <-timeOut:
			log.Info("analyze ctx timeout.")
			return nil
		default:
		}
		AnalyzeAllTables(ctx, tc, db, tables)
	}
}

func TruncateAllTables(ctx context.Context, tc resource.MySQLCompatible, timeout time.Duration, dbName string) error {
	dbUrl := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
	db, err := sql.Open("mysql", resource.BuildDSN(dbUrl, dbName, "timeout=300m", "readTimeout=300m", "writeTimeout=300m"))
	Expect(err).Should(Succeed())
	defer db.Close()
	tables := oltppkg.GetAllTablesNames(ctx, tc, dbName)
	tableNum := len(tables)
	for i := 0; i < tableNum; i++ {
		tableName := tables[i]
		err := oltppkg.DbExecContextWithLog(ctx, db, fmt.Sprintf("truncate table %s", tableName))
		if !errors.Is(err, context.DeadlineExceeded) {
			Expect(err).Should(Succeed())
		}
	}
	return nil
}

func ConnectDbWithTimeout(tc resource.MySQLCompatible, dbName string, params ...string) (*sql.DB, error) {
	dbUrl := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
	db, err := sql.Open("mysql", resource.BuildDSN(dbUrl, dbName, params...))
	if err != nil {
		return nil, err
	}
	return db, nil
}

func DbQueryContextWithLog(ctx context.Context, db *sql.DB, query string) (*sql.Rows, error) {
	log.Info(query)
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		return nil, err
	}
	return rows, nil
}

func DbQueryRowWithLog(db *sql.DB, query string) *sql.Row {
	log.Info(query)
	row := db.QueryRow(query)
	return row
}

func DbExecWithIgnoreCtxDeadline(ctx context.Context, db *sql.DB, execCmd string) error {
	err := oltppkg.DbExecContextWithLog(ctx, db, execCmd)
	if err != nil && !errors.Is(err, context.DeadlineExceeded) {
		return err
	} else {
		return nil
	}
}