package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"
)

var (
	outer *log.Logger
)

func init() {
	fp, _ := os.Create("/tmp/.test.log")
	outer = log.New(fp, "", log.LstdFlags)
}

func LogErr(format string, v ...interface{}) {
	s := fmt.Sprintf(format, v...)
	outer.Output(2, "[ ERR] "+s)
}

func LogInfo(format string, v ...interface{}) {
	if varcfg.LogLevel != string("info") {
		return
	}
	s := fmt.Sprintf(format, v...)
	log.Output(2, "[INFO] "+s)
}

const default_parallel = 8 //最好跟server端可以支持的并发数配置（简单点，就是server端的核心数）.
const default_limit = 30 * 1024

func CheckByRanges(cb checkDbInfo, tbl string, rs []Range, parallel int) (rets []float64, errs []error, err error) {
	dt := cb.dt
	db := cb.db
	if parallel > len(rs) {
		parallel = len(rs)
	}
	LogInfo("CheckByRanges(dt=%v;%s;rs=%d;parallel=%d) staring...\n", dt, tbl, len(rs), parallel)

	rets = make([]float64, len(rs))
	errs = make([]error, len(rs))
	dev, err := stddev_pop(cb, tbl)
	if err != nil {
		LogErr("CheckByRanges.bitXor err:%s", err.Error())
		return nil, nil, err
	}
	LogInfo("CheckByRanges xor=%s.", dev)

	var wg sync.WaitGroup
	var cnt int32
	checkByRanges_worker := func(ch chan int, widx int) {
		for {
			idx, ok := <-ch
			if !ok {
				break
			}
			beg := time.Now()
			//rets[idx], errs[idx] = XorbitByRange(db, tbl, xor, rs[idx])
			rets[idx], errs[idx] = devByRange(db, tbl, dev, rs[idx])
			cost := time.Now().Sub(beg)
			LogInfo("     %s-worker_%d:rs[%d]=%+v(%v). result=%f.err=%v.\n", tbl, widx, idx, rs[idx], cost, rets[idx], errs[idx])
			atomic.AddInt32(&cnt, 1)
		}
		wg.Done()
	}
	ch := make(chan int, parallel)
	wg.Add(parallel)
	for i := 0; i < parallel; i++ {
		go checkByRanges_worker(ch, i)
	}

	//for parallel
	rslen := len(rs)
	for i := 0; i < rslen; i++ {
		ch <- i
	}
	for {
		time.Sleep(time.Microsecond * 100)
		if atomic.LoadInt32(&cnt) == int32(rslen) {
			close(ch)
			break
		}
	}
	wg.Wait()
	LogInfo("CheckByRanges finish....")
	return
}

type checkDbInfo struct {
	db     *sql.DB
	dt     dbType
	dbname string
}

type checkDiffInfo struct {
	pg, tidb checkDbInfo
	parall   int //以慢的db为准.
	tbl      string
	rs       []Range
}

func CheckDiffByRanges(info checkDiffInfo) (errRet []int, skipTable bool) {
	return CheckDiffByRangesWithNotSkip(info, false)
}
func CheckDiffByRangesWithNotSkip(info checkDiffInfo, doNotSkip bool) (errRet []int, skipTable bool) {
	db1, db2 := info.pg, info.tidb
	parall := info.parall
	if parall > len(info.rs) {
		parall = len(info.rs)
	}
	LogInfo("CheckByRanges(%s.%s;rs=%d;%v->%v;parallel=%d) staring...\n", db1.dbname, info.tbl, len(info.rs), db1.dt, db2.dt, parall)
	//get xor;
	dev1, err1 := stddev_pop(db1, info.tbl)
	dev2, err2 := stddev_pop(db2, info.tbl)
	if err1 != nil || err2 != nil {
		LogErr("\t %v->%v; ", db1.dt, err1)
		LogErr("\t %v->%v; ", db2.dt, err2)
		return errRet, skipTable
	}
	LogInfo("%v:%s;", db1.dt, dev1)
	LogInfo("%v:%s;", db2.dt, dev2)
	//start routine;
	var retMtx sync.Mutex
	errRet = make([]int, 0, 128)
	var wg sync.WaitGroup
	var cnt int32
	var errCnt int32
	var ctx, cancel = context.WithCancel(context.TODO())

	checkByRanges_worker := func(ch chan int, widx int, ctx context.Context) {
		defer wg.Done()
		for {
			var idx int
			var ok bool = true
			select {
			case idx, ok = <-ch:
				if !ok {
					return
				}
				atomic.AddInt32(&cnt, 1)
			case <-ctx.Done():
				for ok {
					idx, ok = <-ch
					atomic.AddInt32(&cnt, 1)
				}
				return
			}
			beg := time.Now()
			r := info.rs[idx]
			var ret1, ret2 float64
			var err3, err4 error
			// 这里没有必要用并发增加复杂度（主要耗时在pg，tidb耗时可以忽略）.
			//db 1 check;
			// ret1, err3 = XorbitByRange(db1.db, info.tbl, xor1, r)
			// //db 2 check;
			// ret2, err4 = XorbitByRange(db2.db, info.tbl, xor2, r)
			ret1, err3 = devByRange(db1.db, info.tbl, dev1, r)
			ret2, err4 = devByRange(db2.db, info.tbl, dev2, r)
			//tidb/mysql是有小数点,pg是四舍五入整数了.
			LogInfo("      worker_%d '%s':rs[%d]=%+v(%v).\n", widx, info.tbl, idx, info.rs[idx], time.Now().Sub(beg))
			if err3 != nil || err4 != nil || false == pgEqualTidbDev(ret1, ret2) {
				ec := atomic.AddInt32(&errCnt, 1)
				LogErr("      worker_%d '%s.%s':rs[%d]=%+v(%v).xor(%f-%f) err1=%v;err2=%v.ec=%d\n", widx, db1.dbname, info.tbl, idx, info.rs[idx], time.Now().Sub(beg),
					ret1, ret2, err3, err4, ec)

				if doNotSkip == false && ec > 5 {
					cancel()
					LogErr("'%s.%s' checkDiffDB check too many erros.skip it and do check it latter.\n", db1.dbname, info.tbl)
					retMtx.Lock()
					errRet = nil
					skipTable = true
					retMtx.Unlock()
				} else {
					retMtx.Lock()
					errRet = append(errRet, idx)
					retMtx.Unlock()
				}
			} else {
				//LogInfo("      worker_%d '%s':rs[%d]=%+v(%v).\n", widx, info.tbl, idx, info.rs[idx], time.Now().Sub(beg))
				atomic.StoreInt32(&errCnt, 0)
			}
		}

	}

	ch := make(chan int, parall)
	wg.Add(parall)
	for i := 0; i < parall; i++ {
		go checkByRanges_worker(ch, i, ctx)
	}
	//for parallel
	rslen := len(info.rs)
	for i := 0; i < rslen; i++ {
		ch <- i
	}
	for {
		time.Sleep(time.Microsecond * 100)
		if atomic.LoadInt32(&cnt) == int32(rslen) {
			close(ch)
			break
		}
	}
	wg.Wait()
	LogInfo("CheckDiffByRanges finish....")
	return
}

func pgEqualTidbDev(pgv, tidbv float64) bool {
	tv := (int64(tidbv*10) + 5) / 10
	return int64(pgv) == tv
}