package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"strings"
	"time"
)

const trunck_sql_fmt = `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;`

type ColumnInfo struct {
	Name     string
	Nullable bool
	Type     string
}

type Range struct {
	start, end string
	count      int
}

type tblType string

const (
	base_type tblType = "BASE TABLE"
	view_type tblType = "VIEW"
)

func checkDBPing(dbo *sql.DB) error {
	ctx, cancel := context.WithTimeout(context.TODO(), time.Second*3)
	err := dbo.PingContext(ctx)
	if err != nil {
		cancel()
		return err
	}
	return nil
}

var ErrNoPrimaryKey = fmt.Errorf("table was no primary key")

func GetTableListByDB(cd checkDbInfo, tt tblType) ([]string, error) {
	var query string
	switch cd.dt {
	case db_tidb, db_mysql:
		query = fmt.Sprintf(`SELECT table_name FROM information_schema.tables WHERE table_type = '%s' and table_schema = %s`, string(tt), cd.dbname)
	case db_pg:
		query = `select tablename from pg_tables where schemaname='public'`
	}
	rows, err := cd.db.Query(query)
	if err != nil {
		return nil, err
	}
	var ret []string
	var str string
	for rows.Next() {
		err = rows.Scan(&str)
		if err != nil {
			return nil, err
		}
		ret = append(ret, str)
	}
	return ret, nil
}

func findPrimaryKey(cd checkDbInfo, tbl string) (pk string, err error) {
	switch cd.dt {
	case db_tidb, db_mysql:
		pk, err = findPrimaryKey_tidb(cd, tbl)
	case db_pg:
		pk, err = findPrimaryKey_pg(cd, tbl)
	default:
		return "", fmt.Errorf("findPrimaryKey err: Not support %v", cd.dt)
	}
	if err == sql.ErrNoRows {
		return "", ErrNoPrimaryKey
	}
	return pk, err
}

func getTableColumnInfo(cd checkDbInfo, tbl string) (ret []ColumnInfo, err error) {
	switch cd.dt {
	case db_tidb, db_mysql:
		return getTableColumnInfo_tidb(cd.db, cd.dbname, tbl)
	case db_pg:
		return getTableColumnInfo_pg(cd.db, tbl)
	}
	return nil, fmt.Errorf("getTableColumnInfo err: Not support %v", cd.dt)
}

func query_concat_ws(cols []ColumnInfo, dt dbType) string {
	var names = make([]string, len(cols))
	var isnulls = make([]string, 0, len(cols))
	for idx, col := range cols {
		if dt == db_pg {
			if col.Type == pg_char {
				names[idx] = fmt.Sprintf("trim(%s)", col.Name)
			} else {
				names[idx] = col.Name
			}
		} else {
			names[idx] = col.Name
		}

		if col.Nullable {
			isnulls = append(isnulls, fmt.Sprintf("isnull(%s)", col.Name))
		}
	}
	if len(isnulls) > 0 {
		return fmt.Sprintf("concat_ws(',',%s,concat(%s))", strings.Join(names, ","), strings.Join(isnulls, ","))
	} else {
		return fmt.Sprintf("concat_ws(',',%s)", strings.Join(names, ","))
	}
}

type dbType string

const (
	db_tidb  dbType = "db_type_tidb"
	db_mysql dbType = "db_type_mysql"
	db_pg    dbType = "db_type_postgres"
)

func crc32Type(dt dbType) string {
	switch dt {
	case db_mysql, db_tidb:
		return "UNSIGNED"
	case db_pg:
		return "BIGINT"
	}
	log.Fatalf("not support type '%v'", dt)
	return ""
}

func bitXor(cd checkDbInfo, tbl string) (string, error) {
	cols, err := getTableColumnInfo(cd, tbl)
	if err != nil {
		return "", err
	}
	contract_ws := query_concat_ws(cols, cd.dt)

	return fmt.Sprintf(`bit_xor(CAST(crc32(%s) AS %s))`, contract_ws, crc32Type(cd.dt)), nil
}

//标准差
func stddev_pop(cd checkDbInfo, tbl string) (string, error) {
	cols, err := getTableColumnInfo(cd, tbl)
	if err != nil {
		return "", err
	}
	contract_ws := query_concat_ws(cols, cd.dt)

	return fmt.Sprintf(`stddev_pop(CAST(crc32(%s) AS %s))`, contract_ws, crc32Type(cd.dt)), nil
}

func crc32(cd checkDbInfo, tbl string) (string, error) {
	cols, err := getTableColumnInfo(cd, tbl)
	if err != nil {
		return "", err
	}
	contract_ws := query_concat_ws(cols, cd.dt)

	return fmt.Sprintf(`crc32(%s)`, contract_ws), nil
}

func towhere(tbl ,pk string,r Range)string{
	if r.count <= 0{
		return tbl
	}
	if r.start == "" {
		return fmt.Sprintf("%s where %s <=%s",tbl,pk,r.end)
	}
	return fmt.Sprintf("%s where %s>= %s and %s <=%s",tbl,pk,r.start,pk,r.end)
}
func SplitChunk(db *sql.DB, tbl string, limit int, pk string) ([]Range, error){
	return SplitChunkRange(db,tbl,limit,pk,Range{count: -1})
}

func SplitChunkRange(db *sql.DB, tbl string, limit int, pk string,r Range) ([]Range, error) {
	query := fmt.Sprintf(trunck_sql_fmt, pk, pk,
		pk, pk,
		towhere(tbl,pk,r), limit)
	rows, err := db.Query(query)
	if err != nil {
		return nil, fmt.Errorf("SplitChunkATTidb.Range 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("SplitChunkATTidb.Rows.Scan err:%w", err)
		}
		ret = append(ret, r)
	}
	return ret, nil
}

func MakeQueryBitXor(cd checkDbInfo, tbl string, limit int, pk string) ([]string, error) {
	rs, err := SplitChunk(cd.db, tbl, limit, pk)
	if err != nil {
		return nil, err
	}
	xor, err := bitXor(cd, tbl)
	if err != nil {
		return nil, fmt.Errorf("MakeQueryBitXorAtTable.bitXor err:%s", err)
	}
	var ret = make([]string, len(rs))
	for i, r := range rs {
		ret[i] = fmt.Sprintf(`SELECT %s FROM %s where id >= %s and id <= %s`, xor, tbl, r.start, r.end)
	}
	return ret, nil
}

//一定的几率参数冲突，导致不同的数据产生相同结果.
func XorbitByRange(db *sql.DB, tbl string, xorSql string, r Range) (ret int64, err error) {
	sql := fmt.Sprintf(`SELECT %s FROM %s where id >= %s and id <= %s`, xorSql, tbl, r.start, r.end)
	row := db.QueryRow(sql)
	err = row.Scan(&ret)
	return
}

//标准差 计算.
func devByRange(db *sql.DB, tbl string, devSql string, r Range) (ret float64, err error) {
	sql := fmt.Sprintf(`SELECT %s FROM %s where id >= %s and id <= %s`, devSql, tbl, r.start, r.end)
	row := db.QueryRow(sql)
	err = row.Scan(&ret)
	return
}