// Copyright 2019 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package txn

import (
	"context"
	"database/sql"
	"fmt"
	"math/rand"

	"github.com/pingcap/schrodinger-test/randsql"
	"github.com/pingcap/schrodinger/pkg/test_client"
)

type testSuit struct {
	rg *randsql.RandSQLGen
	// TiDB connection.
	tiDB *sql.DB
	// MySQL connection.
	myDB *sql.DB
}

func NewTestSuit(tidbAddr, tidbPasswd, mysqlAddr, mysqlPasswd, dbName string) *testSuit {
	s := &testSuit{
		rg:   randsql.NewRandSQLGen(),
		tiDB: getDBCli(tidbAddr, tidbPasswd, dbName),
		myDB: getDBCli(mysqlAddr, mysqlPasswd, dbName),
	}
	// See https://internal.pingcap.net/jira/browse/TIDB-4217
	dbMustExec(s.myDB, "set @@global.explicit_defaults_for_timestamp=true;")
	dbMustExec(s.myDB, "set @@explicit_defaults_for_timestamp=true;")
	return s
}

func (s *testSuit) PrepareTables() {
	s.mustCreateTable()
	s.mustCreateTable()
	s.mustCreateTable()
}

func (s *testSuit) StartTest(maxTxnSize int) {
	num := 0
	for {
		err := s.startARoundTransaction(maxTxnSize)
		if err != nil {
			client.Fatalf("err: %#v", err)
		}
		err = s.adminCheckTable()
		if err != nil {
			client.Fatalf("err: %#v", err)
		}
		num++
		if num%5 == 0 {
			err = s.execRandDDL()
			if err != nil {
				client.Fatalf("err: %#v", err)
			}
		}
	}
}

func (s *testSuit) startARoundTransaction(maxTxnSize int) error {
	txnSize := rand.Intn(maxTxnSize) + 1
	ctx := context.Background()
	tiConn, err := s.tiDB.Conn(ctx)
	if err != nil {
		return nil
	}
	defer tiConn.Close()
	myConn, err := s.myDB.Conn(ctx)
	if err != nil {
		return nil
	}
	defer myConn.Close()

	mustExec(ctx, tiConn, "commit")
	mustExec(ctx, myConn, "commit") // make tidb transaction behaviour compatible with mysql.
	mustExec(ctx, tiConn, "set @@tidb_constraint_check_in_place=1;")
	mustExec(ctx, tiConn, "begin")
	mustExec(ctx, myConn, "begin")
	client.Infof("SQL:  %v", "begin")
	for i := 0; i < txnSize; i++ {
		task := s.randDMLSQL()
		if task == nil {
			continue
		}
		err := s.execSQL(ctx, task, tiConn, myConn)
		if err != nil {
			return err
		}
		err = s.checkSelectWithTableName(ctx, tiConn, myConn, task.GetTableName())
		if err != nil {
			return err
		}
	}
	_, tiErr := tiConn.ExecContext(ctx, "commit")
	_, myErr := myConn.ExecContext(ctx, "commit")
	client.Infof("SQL: %v, tidb: %v, mysql: %v", "commit", tiErr, myErr)
	if tiErr == nil && myErr == nil {
		return s.checkRandSelect(ctx, tiConn, myConn)
	}
	if tiErr != nil && myErr != nil && tiErr.Error() == myErr.Error() {
		return s.checkRandSelect(ctx, tiConn, myConn)
	}
	return fmt.Errorf("Error when executing SQL: %s\n remote tidb Err: %#v\n remote mysql err: %#v\n", "commit", tiErr, myErr)
}

func (s *testSuit) randDMLSQL() *randsql.DMLJobTask {
	i := rand.Intn(100)
	if i >= 50 {
		return s.rg.GenRandInsert()
	}
	if i >= 40 {
		return s.rg.GenRandReplace()
	}
	if i >= 5 {
		return s.rg.GenRandUpdate()
	}
	return s.rg.GenRandDelete()
}

func (s *testSuit) execRandDDL() error {
	var task *randsql.DDLJobTask
	for {
		task = s.randDDLSQL()
		if task != nil {
			break
		}
	}
	ctx := context.Background()
	tiConn, err := s.tiDB.Conn(ctx)
	if err != nil {
		return nil
	}
	defer tiConn.Close()
	myConn, err := s.myDB.Conn(ctx)
	if err != nil {
		return nil
	}
	defer myConn.Close()
	mustExec(ctx, myConn, "set @@explicit_defaults_for_timestamp=true;")
	err = s.execSQL(ctx, task, tiConn, myConn)
	if err != nil {
		return err
	}
	return nil
}

func (s *testSuit) randDDLSQL() *randsql.DDLJobTask {
	i := rand.Intn(100)
	if i >= 95 {
		return s.rg.GenRandCreateTable()
	}
	if i >= 90 {
		return s.rg.GenDropTable()
	}
	if i >= 85 {
		return s.rg.GenTruncateTable()
	}
	if i >= 80 {
		return s.rg.GenTruncateTablePartition()
	}
	if i >= 75 {
		return s.rg.GenDropRangePartition()
	}
	if i >= 70 {
		return s.rg.GenAddRangePartition()
	}
	if i >= 60 {
		return s.rg.GenAddIndex()
	}
	if i >= 50 {
		return s.rg.GenDropIndex()
	}
	if i >= 40 {
		return s.rg.GenAddColumn()
	}
	if i >= 25 {
		return s.rg.GenModifyColumn()
	}
	if i >= 5 {
		return s.rg.GenDropColumn()
	}
	return s.rg.GenSetDefaultValue()
}

func (s *testSuit) execSQL(ctx context.Context, task randsql.SQLTask, tiConn, myConn *sql.Conn) error {
	_, tiErr := tiConn.ExecContext(ctx, task.SQL())
	_, myErr := myConn.ExecContext(ctx, task.SQL())
	client.Infof("SQL: %v, tidb: %v, mysql: %v", task.SQL(), tiErr, myErr)
	if tiErr == nil && myErr == nil {
		return s.rg.DoSQLJobInLocal(task, s.tiDB)
	}
	// Not compare error msg because some msg maybe different between tidb and mysql.
	//if tiErr != nil && myErr != nil && tiErr.Error() == myErr.Error() {
	if tiErr != nil && myErr != nil {
		return nil
	}
	return fmt.Errorf("Error when executing SQL: %s\n remote tidb Err: %#v\n remote mysql err: %#v\n%v", task.SQL(), tiErr, myErr, task.Table().DebugPrintToString())
}

func getDBCli(addr, passwd, dbName string) *sql.DB {
	dbDSN := fmt.Sprintf("root:%s@tcp(%s)/%s", passwd, addr, "")
	db, err := sql.Open("mysql", dbDSN)
	if err != nil {
		client.Fatalf("can not connect to database. addr: %s, password: %s", addr, passwd)
	}
	db.SetMaxOpenConns(1)

	_, err = db.Exec("drop database if exists " + dbName)
	assertNil(err)
	_, err = db.Exec("create database " + dbName)
	assertNil(err)
	_, err = db.Exec("use " + dbName)
	assertNil(err)
	client.Infof("connect to database. addr: %s, password: %s, dbName: %v", addr, passwd, dbName)
	return db
}