package main

import (
	"database/sql"
	"fmt"
	"log"
	"strings"
	"sync"
	"testing"

	_ "github.com/go-sql-driver/mysql"
	"github.com/go-yaml/yaml"
	_ "github.com/lib/pq"
	"github.com/stretchr/testify/assert"
)

const (
	pg_host     = "172.16.5.212"
	pg_port     = 54322
	pg_user     = "postgres"
	pg_password = "123456"
	pg_dbname   = "postgres"
)

func newPG() (cd checkDbInfo, er error) {
	psqlInfo := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable", pg_host, pg_port, pg_user, pg_password, pg_dbname)
	db, err := sql.Open("postgres", psqlInfo)
	if err != nil {
		return cd, fmt.Errorf("open db err:%s", err.Error())
	}
	return checkDbInfo{db, db_pg, pg_dbname}, nil
}

func newTidb() (cd checkDbInfo, er error) {
	db, err := sql.Open("mysql", "root:123456@tcp(172.16.4.175:4000)/postgres")
	if err != nil {
		return cd, fmt.Errorf("open db mysql err:%s", err.Error())
	}
	return checkDbInfo{db, db_tidb, pg_dbname}, nil
}

var c_value_template = "###########-###########-###########-###########-###########-###########-###########-###########-###########-###########"

var pad_value_template = "###########-###########-###########-###########-###########"

var base int64

func TestPrepareDB(t *testing.T) {
	d := yaml.NewDecoder(strings.NewReader(testcfg))
	err := d.Decode(&varcfg)
	if err != nil {
		log.Fatalf("parse config error:%s.\nfile=%s.", err.Error(), testcfg)
	}
	//dj,_:=json.MarshalIndent(cfg,"","  ")
	//log.Println("cfg=",string(dj))
	//
	prepareDB(varcfg.PG, varcfg.Tidb, varcfg.Pp)
	return
}

func TestPing(t *testing.T) {
	testPing(t, db_pg)
	testPing(t, db_tidb)
}

func testPing(t *testing.T, dt dbType) {
	var pg checkDbInfo
	var err error
	switch dt {
	case db_pg:
		pg, err = newPG()
	case db_tidb:
		pg, err = newTidb()
	}
	assert.Nil(t, err)
	defer pg.db.Close()
	//err = pg.db.Ping()
	//assert.Nil(t, err)
	//_, err = pg.db.Query("select count(1) from sbtest1")
	//assert.Nil(t, err)
	////
	//qid := 1024
	//row := pg.db.QueryRow("select id from sbtest1 where id = ?", qid)
	//id := 0
	//err = row.Scan(&id)
	//assert.Nil(t, err)
	//assert.Equal(t, qid, id)
	//
	switch dt {
	case db_pg:
		_, err = pg.db.Exec("insert into t values($1,$2),($1,$2)", 1, 1, 2, 2)
		assert.Nil(t, err)
	case db_tidb:
		_, err = pg.db.Exec("insert into t values(?,?),(?,?)", 1, 1, 2, 2)
		assert.Nil(t, err)
	}
}

func checkTimes(tbl string, limit int64, db *sql.DB) (int64, error) {
	query := fmt.Sprintf(`select count(1) from %s`, tbl)
	row := db.QueryRow(query)
	var cnt int64
	err := row.Scan(&cnt)
	if err != nil {
		return 0, err
	}
	return (cnt + limit - 1) / limit, nil
}

const trunck_sql = `select min(t.%s) as start_key, max(t.%s) as end_key, count(*) as page_size from 
    (
        select %s, row_number() over(order by %s) as row_num 
            from %s
    ) t 
    group by floor((t.row_num-1)/%d) order by start_key;`

func splitBlock(tbl string, pkName string, limit int64, db *sql.DB) ([]Range, error) {
	query := fmt.Sprintf(trunck_sql, pkName, pkName, pkName, pkName, tbl, limit)
	rows, err := db.Query(query)
	if err != nil {
		return nil, fmt.Errorf("splitBlock err:%w", err)
	}
	var ret []Range
	for rows.Next() {
		var r Range
		err = rows.Scan(&r.start, &r.end, &r.count)
		if err != nil {
			return nil, fmt.Errorf("splitBlock err2:%w", err)
		}
		ret = append(ret, r)
	}
	return ret, nil
}

func checkSum(tbl string, r Range, db *sql.DB) (cksum int64, err error) {
	query := `SELECT bit_xor(
    CAST(crc32(
        concat_ws(',',
            id,k,c,pad
        )
    ) AS BIGINT)
)
FROM %s where id >= %s and id <= %s`
	row := db.QueryRow(fmt.Sprintf(query, tbl, r.start, r.end))
	err = row.Scan(&cksum)
	return
}

func check(tbl string, pkName string, limit int64, db *sql.DB) error {
	ranges, err := splitBlock(tbl, pkName, limit, db)
	if err != nil {
		return err
	}
	fmt.Printf("ranges start(%+v) length=%d.\n", ranges[0], len(ranges))
	var wg sync.WaitGroup
	for idx, r := range ranges {
		if idx >= 10 {
			break
		}
		wg.Add(1)
		go func(rv Range) {
			ck, err := checkSum(tbl, rv, db)
			if err != nil {
				fmt.Println("checkSum err:", err)
				return
			}
			fmt.Println("checkSum range=", rv, ";checksum=", ck)
			wg.Done()
		}(r)
	}
	wg.Wait()
	return nil
}

func TestCheck(t *testing.T) {
	tbl := "sbtest1"
	pkName := "id"
	limit := int64(30000)
	pg, err := newPG()
	defer pg.db.Close()
	assert.Nil(t, err)
	err = check(tbl, pkName, limit, pg.db)
	assert.Nil(t, err)
}

func TestYaml(t *testing.T) {
	in := testcfg
	var cfg Cfg
	err := yaml.Unmarshal([]byte(in), &cfg)
	assert.Nil(t, err)
	t.Logf("%+v\n", cfg)
}

var (
	db     = "postgres"
	tables = []string{"sbtest1", "t", "t1"}
)

func TestGetColInfo_pg(t *testing.T) {
	pg, _ := newPG()
	testGetColInfo(t, pg)
	pg.db.Close()
}
func TestGetColInfo_tidb(t *testing.T) {
	tidb, _ := newTidb()
	testGetColInfo(t, tidb)
	tidb.db.Close()
}

func testGetColInfo(t *testing.T, info checkDbInfo) {
	for _, tbl := range []string{"sbtest0", "t", "t1"} {
		cols, err := getTableColumnInfo(info, tbl)
		assert.Nil(t, err)
		t.Log(tbl, ":")
		for idx, col := range cols {
			if col.Nullable {
				t.Logf("  %d:%s-Nullable", idx, col.Name)
			} else {
				t.Logf("  %d:%s-not null", idx, col.Name)
			}
		}
	}
}

func TestFindPrimaryKey_pg(t *testing.T) {
	pg, _ := newPG()
	testFindPrimaryKey(t, pg)
	pg.db.Close()
}

func TestFindPrimaryKey_tidb(t *testing.T) {
	tidb, _ := newTidb()
	testFindPrimaryKey(t, tidb)
	tidb.db.Close()
}

func testFindPrimaryKey(t *testing.T, cd checkDbInfo) {
	//
	for _, tbl := range []string{"sbtest1", "t", "t1"} {
		pk, err := findPrimaryKey(cd, tbl)
		if err == ErrNoPrimaryKey {
			t.Errorf("%s was no primary key", tbl)
		} else {
			assert.Nil(t, err)
			t.Logf("%s.%s", tbl, pk)
		}

	}
}

func TestSplitChunk(t *testing.T) {
	pg, _ := newPG()
	tidb, _ := newTidb()
	rs_pg := testSplitChunk(t, pg)
	rs_tidb := testSplitChunk(t, tidb)
	assert.Equal(t, len(rs_pg), len(rs_tidb))
	for i, r := range rs_pg {
		rt := rs_tidb[i]
		assert.Equal(t, r.start, rt.start)
		assert.Equal(t, r.end, rt.end)
		assert.Equal(t, r.count, rt.count)
	}
}

func testSplitChunk(t *testing.T, cd checkDbInfo) []Range {
	//
	tbl := "sbtest1"
	pk, err := findPrimaryKey(cd, tbl)
	assert.Nil(t, err)
	rs, err := SplitChunk(cd.db, tbl, 654321, pk)
	assert.Nil(t, err)
	return rs
}

func TestCheckByRanges_tidb(t *testing.T) {

	tidb, _ := newTidb()
	testCheckByRanges(t, tidb, 654321)
	tidb.db.Close()
}

func TestCheckByRanges_pg(t *testing.T) {
	pg, _ := newPG()
	testCheckByRanges(t, pg, 54321)
	pg.db.Close()
}

func testCheckByRanges(t *testing.T, cd checkDbInfo, limit int) {
	//
	tbl := "sbtest1"
	pk, err := findPrimaryKey(cd, tbl)
	assert.Nil(t, err)
	rs, err := SplitChunk(cd.db, tbl, limit, pk)
	assert.Nil(t, err)
	//
	rets, errs, err := CheckByRanges(cd, tbl, rs, 8)
	assert.Nil(t, err)
	for i := 0; i < len(rs); i++ {
		assert.Nil(t, errs[i])
		t.Logf("%d: %v --> %f.", i, rs[i], rets[i])
	}
}

func TestCheckDiffByRanges(t *testing.T) {
	pg, err := newPG()
	assert.Nil(t, err)
	defer pg.db.Close()
	tidb, err := newTidb()
	assert.Nil(t, err)
	defer tidb.db.Close()
	//
	tbl := "sbtest0"
	limit := 1 * 1000
	pk, err := findPrimaryKey(tidb, tbl)
	assert.Nil(t, err)
	rs, err := SplitChunk(tidb.db, tbl, limit, pk)
	assert.Nil(t, err)
	//
	cdi := checkDiffInfo{
		pg:     pg,
		tidb:   tidb,
		tbl:    tbl,
		parall: 8,
		rs:     rs,
	}
	CheckDiffByRanges(cdi)
}

func TestDoCheckDiff(t *testing.T) {
	d := yaml.NewDecoder(strings.NewReader(testcfg))
	err := d.Decode(&varcfg)
	if err != nil {
		log.Fatalf("parse config error:%s.\nfile=%s.", err.Error(), testcfg)
	}
	doCheckDiff(varcfg.PG, varcfg.Tidb, varcfg.CD)
}

func initConfig() (checkDiffInfo, checkRange) {
	d := yaml.NewDecoder(strings.NewReader(testcfg))
	err := d.Decode(&varcfg)
	if err != nil {
		log.Fatalf("parse config error:%s.\nfile=%s.", err.Error(), testcfg)
	}
	tidb, err := makeTidbDatabase(varcfg.Tidb, db)
	if err != nil {
		log.Fatalf(err.Error())
	}
	pg, err := makePGDatabase(varcfg.PG, db)
	if err != nil {
		log.Fatalf(err.Error())
	}
	var cdr = varcfg.CR
	return checkDiffInfo{
		pg:     pg,
		tidb:   tidb,
		parall: cdr.NThread,
		tbl:    cdr.Tbl,
	}, cdr
}

func TestReCheckRange(t *testing.T) {
	cdi, cdr := initConfig()
	for _, cr := range cdr.Ranges {
		r := Range{
			cr.Start,
			cr.End,
			cdr.Count,
		}
		tbl := reCheckRange(cdi, r, "id")
		t.Logf("%+v", tbl)
	}
}

func TestLoopCheckRanges(t *testing.T) {
	cdi, cdr := initConfig()
	var rs []Range
	for _, cr := range cdr.Ranges {
		rs = append(rs, Range{
			cr.Start,
			cr.End,
			cdr.Count,
		})
	}
	tblSum, ids := loopCheckRanges(cdi, rs)
	t.Logf("tblSum=%+v", tblSum)
	var b strings.Builder
	if len(ids) > 0 {
		showIds(ids, &b, "")
	}
	t.Logf("ids=%s", b.String())
}

func TestXor(t *testing.T) {
	rets1 := []uint32{2562743114, 3323869370, 2238977323, 250335355} // 08 ; 3577849504
	rets2 := []uint32{1416256468, 179583012, 1239291317, 1503405136} // 09 ;  1318867477
	xv1 := myXor(rets1)
	xv2 := myXor(rets2)
	assert.Equal(t, int64(xv1), int64(xv2))
}

func myXor(vs []uint32) (ret uint32) {
	ret = vs[0]
	for _, v := range vs[1:] {
		ret = ret ^ v
	}
	return
}