package sqltest

import (
	"database/sql"
	"flag"
	"fmt"
	"io/ioutil"
	"math/rand"
	"net"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"testing"

	_ "github.com/go-sql-driver/mysql"
	"github.com/pingcap/tidb-test/util"
	"github.com/pingcap/tidb/session"
)

var (
	store = flag.String("store", "memory", "registered store name, [memory, goleveldb, boltdb, tikv]")
	path  = flag.String("path", "test/%s", "store path")
)

type Tester interface {
	RunTest(*testing.T, func(params))
}

var (
	myMysql Tester = &myMysqlDB{}
	goMysql Tester = &goMysqlDB{}
	sqlite  Tester = sqliteDB{}
	pq      Tester = &pqDB{}
	oracle  Tester = &oracleDB{}
	ti      Tester = &tiDB{}
)

const TablePrefix = "gosqltest_"

// pqDB validates the postgres driver by Blake Mizerany (github.com/bmizerany/pq.go)
type pqDB struct {
	once    sync.Once // guards init of running
	running bool      // whether port 5432 is listening
}

func (p *pqDB) RunTest(t *testing.T, fn func(params)) {
	if !p.Running() {
		fmt.Printf("skipping test; no Postgres running on localhost:5432\n")
		return
	}
	user := os.Getenv("GOSQLTEST_PQ_USER")
	if user == "" {
		user = os.Getenv("USER")
	}
	dbName := "gosqltest"
	db, err := sql.Open("postgres", fmt.Sprintf("user=%s password=gosqltest dbname=%s sslmode=disable", user, dbName))
	if err != nil {
		t.Fatalf("error connecting: %v", err)
	}

	params := params{pq, t, db}

	// Drop all tables in the test database.
	rows, err := db.Query("SELECT table_name FROM information_schema.tables WHERE table_name LIKE '" +
		TablePrefix + "%' AND table_schema = 'public'")
	if err != nil {
		t.Fatalf("failed to enumerate tables: %v", err)
	}
	for rows.Next() {
		var table string
		if rows.Scan(&table) == nil {
			params.mustExec("DROP TABLE " + table)
		}
	}

	fn(params)
}

func (p *pqDB) Running() bool {
	p.once.Do(func() {
		c, err := net.Dial("tcp", "localhost:5432")
		if err == nil {
			p.running = true
			c.Close()
		}
	})
	return p.running
}

type sqliteDB struct{}

type myMysqlDB struct {
	once    sync.Once // guards init of running
	running bool      // whether port 3306 is listening
}

func (m *myMysqlDB) Running() bool {
	m.once.Do(func() {
		c, err := net.Dial("tcp", "localhost:3306")
		if err == nil {
			m.running = true
			c.Close()
		}
	})
	return m.running
}

type goMysqlDB struct {
	once    sync.Once // guards init of running
	running bool      // whether port 3306 is listening
}

func (m *goMysqlDB) Running() bool {
	m.once.Do(func() {
		c, err := net.Dial("tcp", "localhost:3306")
		if err == nil {
			m.running = true
			c.Close()
		}
	})
	return m.running
}

type oracleDB struct {
	once    sync.Once // guards init of running
	running bool      // whether port 1521 is listening
}

func (o *oracleDB) Running() bool {
	o.once.Do(func() {
		c, err := net.Dial("tcp", "localhost:1521")
		if err == nil {
			o.running = true
			c.Close()
		}
	})
	return o.running
}

type tiDB struct {
	once    sync.Once // guards init of running
	running bool      // whether port 1521 is listening
}

func (o *tiDB) Running() bool {
	o.running = true
	return o.running
}

type params struct {
	dbType Tester
	*testing.T
	*sql.DB
}

func (t params) mustExec(sql string, args ...interface{}) sql.Result {
	res, err := t.DB.Exec(sql, args...)
	if err != nil {
		t.Fatalf("Error running %q: %v", sql, err)
	}
	return res
}

var qrx = regexp.MustCompile(`\?`)

// q converts "?" characters to $1, $2, $n on postgres, :1, :2, :n on Oracle
func (t params) q(sql string) string {
	var pref string
	switch t.dbType {
	case pq:
		pref = "$"
	case oracle:
		pref = ":"
	default:
		return sql
	}
	n := 0
	return qrx.ReplaceAllStringFunc(sql, func(string) string {
		n++
		return pref + strconv.Itoa(n)
	})
}

func (m sqliteDB) RunTest(t *testing.T, fn func(params)) {
	tempDir, err := ioutil.TempDir("", "")
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(tempDir)
	db, err := sql.Open("sqlite3", filepath.Join(tempDir, "foo.db"))
	if err != nil {
		t.Fatalf("foo.db open fail: %v", err)
	}
	fn(params{sqlite, t, db})
}

func (m *myMysqlDB) RunTest(t *testing.T, fn func(params)) {
	if !m.Running() {
		t.Logf("skipping test; no MySQL running on localhost:3306")
		return
	}
	user := os.Getenv("GOSQLTEST_MYSQL_USER")
	if user == "" {
		user = "root"
	}
	pass, ok := getenvOk("GOSQLTEST_MYSQL_PASS")
	if !ok {
		pass = "root"
	}
	dbName := "gosqltest"
	db, err := sql.Open("mymysql", fmt.Sprintf("%s/%s/%s", dbName, user, pass))
	if err != nil {
		t.Fatalf("error connecting: %v", err)
	}

	params := params{myMysql, t, db}

	// Drop all tables in the test database.
	rows, err := db.Query("SHOW TABLES")
	if err != nil {
		t.Fatalf("failed to enumerate tables: %v", err)
	}
	for rows.Next() {
		var table string
		if rows.Scan(&table) == nil &&
			strings.HasPrefix(strings.ToLower(table), strings.ToLower(TablePrefix)) {
			params.mustExec("DROP TABLE " + table)
		}
	}

	fn(params)
}

func (m *goMysqlDB) RunTest(t *testing.T, fn func(params)) {
	if !m.Running() {
		t.Logf("skipping test; no MySQL running on localhost:3306")
		return
	}
	user := os.Getenv("GOSQLTEST_MYSQL_USER")
	if user == "" {
		user = "root"
	}
	pass, ok := getenvOk("GOSQLTEST_MYSQL_PASS")
	if !ok {
		pass = "root"
	}
	dbName := "gosqltest"
	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@/%s", user, pass, dbName))
	if err != nil {
		t.Fatalf("error connecting: %v", err)
	}

	params := params{goMysql, t, db}

	// Drop all tables in the test database.
	rows, err := db.Query("SHOW TABLES")
	if err != nil {
		t.Fatalf("failed to enumerate tables: %v", err)
	}
	for rows.Next() {
		var table string
		if rows.Scan(&table) == nil &&
			strings.HasPrefix(strings.ToLower(table), strings.ToLower(TablePrefix)) {
			params.mustExec("DROP TABLE " + table)
		}
	}

	fn(params)
}

func (o *oracleDB) RunTest(t *testing.T, fn func(params)) {
	if !o.Running() {
		t.Logf("skipping test; no Oracle running on localhost:1521")
		return
	}
	db, err := sql.Open("goracle", os.Getenv("GOSQLTEST_ORACLE"))
	if err != nil {
		t.Fatalf("error connecting: %v", err)
	}
	defer db.Close()

	params := params{oracle, t, db}

	// Drop all tables in the test database.
	rows, err := db.Query(`SELECT table_name FROM user_tables
        WHERE UPPER(table_name) LIKE UPPER('` + TablePrefix + `%')`)
	if err != nil {
		t.Fatalf("failed to enumerate tables: %v", err)
	}
	var table sql.NullString
	for rows.Next() {
		err = rows.Scan(&table)
		// t.Logf("Next => table=%+v err=%s", table, err)
		if err != nil {
			t.Fatalf("error reading table name: %s", err)
		} else if !table.Valid {
			t.Fatalf("error reading table name: null value!")
		} else {
			params.mustExec("DROP TABLE " + table.String)
		}
	}

	fn(params)
}

func (o *tiDB) RunTest(t *testing.T, fn func(params)) {
	if !o.Running() {
		t.Logf("skipping test; no TiDB")
		return
	}
	flag.Parse()
	session.SetSchemaLease(0)
	dbName := "test"
	db, err := util.OpenDBWithRetry("mysql", "root@tcp(localhost:4001)/"+dbName+"?strict=true")
	if err != nil {
		t.Fatalf("error connecting: %v", err)
	}

	params := params{ti, t, db}

	// Drop all tables in the test database.
	rows, err := db.Query("SHOW TABLES")
	if err != nil {
		t.Fatalf("failed to enumerate tables: %v", err)
	}
	for rows.Next() {
		var table string
		if rows.Scan(&table) == nil &&
			strings.HasPrefix(strings.ToLower(table), strings.ToLower(TablePrefix)) {
			params.mustExec("DROP TABLE " + table)
		}
	}
	fn(params)
}

func sqlBlobParam(t params, size int) string {
	switch t.dbType {
	case sqlite:
		return fmt.Sprintf("blob[%d]", size)
	case pq:
		return "bytea"
	case oracle:
		return fmt.Sprintf("RAW(%d)", size)
	}
	return fmt.Sprintf("VARBINARY(%d)", size)
}

// func TestBlobs_SQLite(t *testing.T)  { sqlite.RunTest(t, testBlobs) }
// func TestBlobs_MyMySQL(t *testing.T) { myMysql.RunTest(t, testBlobs) }
// func TestBlobs_GoMySQL(t *testing.T) { goMysql.RunTest(t, testBlobs) }
// func TestBlobs_PQ(t *testing.T)      { pq.RunTest(t, testBlobs) }
// func TestBlobs_Oracle(t *testing.T)  { oracle.RunTest(t, testBlobs) }
func TestBlobs_TiDB(t *testing.T) { ti.RunTest(t, testBlobs) }

func testBlobs(t params) {
	var blob = []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}
	t.mustExec("create table " + TablePrefix + "foo (id integer primary key, bar " + sqlBlobParam(t, 16) + ")")
	t.mustExec(t.q("insert into "+TablePrefix+"foo (id, bar) values(?,?)"), 0, blob)

	want := fmt.Sprintf("%x", blob)

	b := make([]byte, 16)
	err := t.QueryRow(t.q("select bar from "+TablePrefix+"foo where id = ?"), 0).Scan(&b)
	got := fmt.Sprintf("%x", b)
	if err != nil {
		t.Errorf("[]byte scan: %v", err)
	} else if got != want {
		t.Errorf("for []byte, got %q; want %q", got, want)
	}

	err = t.QueryRow(t.q("select bar from "+TablePrefix+"foo where id = ?"), 0).Scan(&got)
	want = string(blob)
	if err != nil {
		t.Errorf("string scan: %v", err)
	} else if got != want {
		t.Errorf("for string, got %q; want %q", got, want)
	}
}

// func TestManyQueryRow_SQLite(t *testing.T)  { sqlite.RunTest(t, testManyQueryRow) }
// func TestManyQueryRow_MyMySQL(t *testing.T) { myMysql.RunTest(t, testManyQueryRow) }
// func TestManyQueryRow_GoMySQL(t *testing.T) { goMysql.RunTest(t, testManyQueryRow) }
// func TestManyQueryRow_PQ(t *testing.T)      { pq.RunTest(t, testManyQueryRow) }
// func TestManyQueryRow_Oracle(t *testing.T)  { oracle.RunTest(t, testManyQueryRow) }
func TestManyQueryRow_TiDB(t *testing.T) { ti.RunTest(t, testManyQueryRow) }

func testManyQueryRow(t params) {
	if testing.Short() {
		t.Logf("skipping in short mode")
		return
	}
	t.mustExec("create table " + TablePrefix + "foo (id integer primary key, name varchar(50))")
	t.mustExec(t.q("insert into "+TablePrefix+"foo (id, name) values(?,?)"), 1, "bob")
	var name string
	for i := 0; i < 10000; i++ {
		err := t.QueryRow(t.q("select name from "+TablePrefix+"foo where id = ?"), 1).Scan(&name)
		if err != nil || name != "bob" {
			t.Fatalf("on query %d: err=%v, name=%q", i, err, name)
		}
	}
}

// func TestTxQuery_SQLite(t *testing.T)  { sqlite.RunTest(t, testTxQuery) }
// func TestTxQuery_MyMySQL(t *testing.T) { myMysql.RunTest(t, testTxQuery) }
// func TestTxQuery_GoMySQL(t *testing.T) { goMysql.RunTest(t, testTxQuery) }
// func TestTxQuery_PQ(t *testing.T)      { pq.RunTest(t, testTxQuery) }
// func TestTxQuery_Oracle(t *testing.T)  { oracle.RunTest(t, testTxQuery) }
func TestTxQuery_TiDB(t *testing.T) { ti.RunTest(t, testTxQuery) }

func testTxQuery(t params) {
	tx, err := t.Begin()
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback()

	tableName := TablePrefix + "foo"

	_, err = tx.Exec("create table " + tableName + " (id integer primary key, name varchar(50))")
	if err != nil {
		t.Logf("cannot create table "+tableName+": %s", err)
	}

	_, err = tx.Exec(t.q("insert into "+tableName+" (id, name) values(?,?)"), 1, "bob")
	if err != nil {
		t.Fatal(err)
	}

	r, err := tx.Query(t.q("select name from "+tableName+" where id = ?"), 1)
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()

	if !r.Next() {
		if r.Err() != nil {
			t.Fatal(err)
		}
		t.Fatal("expected one rows")
	}

	var name string
	err = r.Scan(&name)
	if err != nil {
		t.Fatal(err)
	}
}

// func TestPreparedStmt_SQLite(t *testing.T)  { sqlite.RunTest(t, testPreparedStmt) }
// func TestPreparedStmt_MyMySQL(t *testing.T) { myMysql.RunTest(t, testPreparedStmt) }
// func TestPreparedStmt_GoMySQL(t *testing.T) { goMysql.RunTest(t, testPreparedStmt) }
// func TestPreparedStmt_PQ(t *testing.T)      { pq.RunTest(t, testPreparedStmt) }
// func TestPreparedStmt_Oracle(t *testing.T)  { oracle.RunTest(t, testPreparedStmt) }
func TestPreparedStmt_TiDB(t *testing.T) { ti.RunTest(t, testPreparedStmt) }

func testPreparedStmt(t params) {
	t.mustExec("CREATE TABLE " + TablePrefix + "t (count INT)")
	sel, err := t.Prepare("SELECT count FROM " + TablePrefix + "t ORDER BY count DESC")
	if err != nil {
		t.Fatalf("prepare 1: %v", err)
	}
	ins, err := t.Prepare(t.q("INSERT INTO " + TablePrefix + "t (count) VALUES (?)"))
	if err != nil {
		t.Fatalf("prepare 2: %v", err)
	}

	for n := 1; n <= 3; n++ {
		if _, err := ins.Exec(n); err != nil {
			t.Fatalf("insert(%d) = %v", n, err)
		}
	}

	const nRuns = 10
	ch := make(chan bool)
	for i := 0; i < nRuns; i++ {
		go func() {
			defer func() {
				ch <- true
			}()
			for j := 0; j < 10; j++ {
				count := 0
				if err := sel.QueryRow().Scan(&count); err != nil && err != sql.ErrNoRows {
					t.Errorf("Query: %v", err)
					return
				}
				if _, err := ins.Exec(rand.Intn(100)); err != nil {
					t.Errorf("Insert: %v", err)
					return
				}
			}
		}()
	}
	for i := 0; i < nRuns; i++ {
		<-ch
	}
}

// This test already exists in TiDB, but add it here to cover coprocessor in tikv.
func TestTimestampTimeZone(t *testing.T) { ti.RunTest(t, testTimestampTimeZone) }

func testTimestampTimeZone(t params) {
	t.mustExec("use test")
	t.Exec("drop table if exists t")
	t.mustExec("create table t (ts timestamp)")
	tx0, err := t.Begin()
	if err != nil {
		t.Fatal(err)
	}
	func() {
		_, err = tx0.Exec("set time_zone = '+00:00'")
		if err != nil {
			t.Error(err)
			return
		}
		_, err = tx0.Exec("insert into t values ('2017-04-27 22:40:42')")
		if err != nil {
			t.Error(err)
			return
		}
		err = tx0.Commit()
		if err != nil {
			t.Error(err)
			return
		}
	}()
	// The timestamp will get different value if time_zone session variable changes.
	tests := []struct {
		timezone string
		expect   string
	}{
		{"+10:00", "2017-04-28 08:40:42"},
		{"-6:00", "2017-04-27 16:40:42"},
	}
	for _, tt := range tests {
		func() {
			tx, err := t.Begin()
			if err != nil {
				t.Fatal(err)
			}
			defer tx.Rollback()
			_, err = tx.Exec(fmt.Sprintf("set time_zone = '%s'", tt.timezone))
			if err != nil {
				t.Error(err)
				return
			}
			rows, err := tx.Query(fmt.Sprintf("select * from t where ts = '%s'", tt.expect))
			if err != nil {
				t.Error(err)
				return
			}
			if !rows.Next() {
				t.Errorf("%s not found", tt.expect)
				return
			}
			var str string
			err = rows.Scan(&str)
			if err != nil {
				t.Error(err)
			}
			if str != tt.expect {
				t.Errorf("expect: %s, get: %s", tt.expect, str)
			}
		}()
	}

	// For issue https://github.com/pingcap/tidb/issues/3467
	t.Exec("drop table if exists t1")
	t.mustExec(`CREATE TABLE t1 (
	      id bigint(20) NOT NULL AUTO_INCREMENT,
	      uid int(11) DEFAULT NULL,
	      datetime datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
	      ip varchar(128) DEFAULT NULL,
	    PRIMARY KEY (id),
	      KEY i_datetime (datetime),
	      KEY i_userid (uid)
	    );`)
	t.mustExec(`INSERT INTO t1 VALUES (123381351,1734,"2014-03-31 08:57:10","127.0.0.1");`)

	var result1 string // Cover TableReaderExec
	t.mustQuery("select datetime from t1;", &result1)
	t.check(result1, "2014-03-31 08:57:10")

	var result2 string // Cover IndexReaderExec
	t.mustQuery("select datetime from t1 where datetime='2014-03-31 08:57:10'", &result2)
	t.check(result2, "2014-03-31 08:57:10")

	var (
		a       int
		b       int
		result3 string
		d       string
	)
	// Cover IndexLookupExec
	t.mustQuery("select * from t1 where datetime='2014-03-31 08:57:10'", &a, &b, &result3, &d)
	t.check(result3, "2014-03-31 08:57:10")
}

func (t params) mustQuery(sql string, values ...interface{}) {
	rows, err := t.Query(sql)
	if err != nil {
		t.Error(err)
		return
	}

	rows.Next()
	err = rows.Scan(values...)
	if err != nil {
		t.Error(err)
	}
}

func (t params) check(str1, str2 string) {
	if str1 != str2 {
		t.Errorf("expect: %s, get: %s\n", str2, str1)
	}
}

func getenvOk(k string) (v string, ok bool) {
	v = os.Getenv(k)
	if v != "" {
		return v, true
	}
	keq := k + "="
	for _, kv := range os.Environ() {
		if kv == keq {
			return "", true
		}
	}
	return "", false
}