package tpcc

import (
	"bytes"
	"database/sql"
	"fmt"
	"io/ioutil"
	"os/exec"
	"strconv"
	"strings"

	_ "github.com/go-sql-driver/mysql"
	"github.com/juju/errors"
	"github.com/pingcap/schrodinger/pkg/test_client"
)

func NewTPCCCase(inputDSN, outputDSN string, warehouse int) *TPCCCase {
	return &TPCCCase{
		inputDSN:  inputDSN,
		outputDSN: outputDSN,
		warehouse: warehouse,
	}
}

type TPCCCase struct {
	inputDSN  string
	outputDSN string
	warehouse int
}

func (t *TPCCCase) Execute() error {
	client.Infof("starting tpcc test at %+v", t)
	defer func() {
		client.Infof("finishing tpcc test at %+v", t)
	}()
	return errors.Trace(t.runTest())
}

func (t *TPCCCase) runTest() error {
	if err := t.createNewDB(); err != nil {
		return errors.Trace(err)
	}
	runPathName := "/benchmarksql50-mysql/run"
	propsFileName := "props.mysql"

	if err := t.modifyPropsFile(fmt.Sprintf("%s/%s", runPathName, propsFileName)); err != nil {
		return errors.Trace(err)
	}
	if err := t.runCMD(exec.Command("./runSQL.sh", propsFileName, "tableCreates"), runPathName); err != nil {
		return errors.Trace(err)
	}
	if err := t.runCMD(exec.Command("./runSQL.sh", propsFileName, "indexCreates"), runPathName); err != nil {
		return errors.Trace(err)
	}
	if err := t.runCMD(exec.Command("./runLoader.sh", propsFileName), runPathName); err != nil {
		return errors.Trace(err)
	}
	if err := t.runCMD(exec.Command("./runBenchmark.sh", propsFileName), runPathName); err != nil {
		return errors.Trace(err)
	}
	if err := t.checkResults(); err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (t *TPCCCase) modifyPropsFile(filename string) error {
	if data, err := ioutil.ReadFile(filename); err != nil {
		return errors.Trace(err)
	} else {
		data = []byte(strings.Replace(string(data), "DSN", t.inputDSN, 1))
		data = []byte(strings.Replace(string(data), "WH", strconv.Itoa(t.warehouse), 1))
		if err := ioutil.WriteFile(filename, data, 0644); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (t *TPCCCase) runCMD(cmd *exec.Cmd, dir string) error {
	cmd.Dir = dir
	var stdErr bytes.Buffer
	cmd.Stderr = &stdErr
	cmdStr := fmt.Sprintf("%s", strings.Join(cmd.Args, " "))
	client.Infof("running shell command: %s", cmdStr)
	if err := cmd.Run(); err != nil {
		return errors.Annotatef(err, "cmd: %s failed\nerr: %s\nstderr: %s", cmdStr, err, stdErr.String())
	}
	return nil
}

// We re-create the database before every test case start to avoid the influence of the dirty data
func (t *TPCCCase) createNewDB() error {
	client.Info("creating a new database")
	tidbServer, err := sql.Open("mysql", fmt.Sprintf("root:@tcp(%s)/", t.inputDSN))
	if err != nil {
		return errors.Trace(err)
	}
	defer tidbServer.Close()
	_, err = tidbServer.Exec("drop database if exists tpcc")
	if err != nil {
		return errors.Trace(err)
	}
	_, err = tidbServer.Exec("create database tpcc")
	if err != nil {
		return errors.Trace(err)
	}
	_, err = tidbServer.Exec("set global tidb_opt_insubq_to_join_and_agg = 1")
	if err != nil {
		return errors.Trace(err)
	}
	client.Info("new database is created")
	return nil
}

func (t *TPCCCase) checkResults() error {
	tidbServer, err := sql.Open("mysql", fmt.Sprintf("root:@tcp(%s)/%s", t.outputDSN, "tpcc"))
	if err != nil {
		return errors.Trace(err)
	}
	defer tidbServer.Close()
	client.Info("checking data size start!")
	if _, err := tidbServer.Exec("update mysql.tidb set VARIABLE_VALUE='600m' where  VARIABLE_NAME='tikv_gc_life_time'"); err != nil {
		return errors.Trace(err)
	}
	defer func() {
		if _, err := tidbServer.Exec("update mysql.tidb set VARIABLE_VALUE='10m' where  VARIABLE_NAME='tikv_gc_life_time'"); err != nil {
			client.Info("reset gc time error: ", err)
		}
	}()
	if err := t.checkData(tidbServer); err != nil {
		client.Info("checking data size failed!")
		return errors.Trace(err)
	}
	client.Info("checking data size finish!")
	client.Info("checking consistency start!")
	if err := t.checkConsistency(tidbServer); err != nil {
		client.Info("checking consistency failed!")
		return errors.Trace(err)
	}
	client.Info("checking consistency finish!")
	// TODO: add the admin check table back when the performance of it is improved.
	// client.Info("checking table start!")
	//if err := t.checkTables(tidbServer); err != nil {
	// 	client.Info("checking table failed!")
	//	return errors.Trace(err)
	//}
	// client.Info("checking table finish!")
	return nil
}

type tableSize struct {
	name string
	size int
}

func (t *TPCCCase) checkData(tidbServer *sql.DB) error {
	tableSizes := []tableSize{
		{"bmsql_warehouse", t.warehouse},
		{"bmsql_district", 10 * t.warehouse},
		{"bmsql_customer", 30000 * t.warehouse},
		{"bmsql_item", 100000},
		{"bmsql_stock", 100000 * t.warehouse},
	}
	for _, v := range tableSizes {
		if err := t.checkDataSize(tidbServer, v); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (t *TPCCCase) checkDataSize(tidbServer *sql.DB, tableSize tableSize) error {
	rs, err := tidbServer.Query("select count(*) from " + tableSize.name)
	if err != nil {
		return errors.Trace(err)
	}
	var out int
	if !rs.Next() {
		return errors.Errorf("%s NO DATA FOUND!", tableSize.name)
	}
	err = rs.Scan(&out)
	if err != nil {
		return errors.Trace(err)
	}
	if out != tableSize.size {
		return errors.Errorf("%s DATA SIZE NOT RIGHT! NEED %d BUT GOT %d.",
			tableSize.name, tableSize.size, out)
	}
	return nil
}

func (t *TPCCCase) checkConsistency(tidbServer *sql.DB) error {
	for i, sqlStr := range consistencyCheckSQLs {
		rs, err := tidbServer.Query(sqlStr)
		if err != nil {
			return errors.Trace(err)
		}
		if rs.Next() {
			return errors.Errorf("DATA NOT CONSISTENCY! SQL%d returned data!", i)
		}
	}
	return nil
}

func (t *TPCCCase) checkTables(tidbServer *sql.DB) error {
	tableNames := []string{
		"bmsql_config",
		"bmsql_customer",
		"bmsql_district",
		"bmsql_history",
		"bmsql_item",
		"bmsql_new_order",
		"bmsql_oorder",
		"bmsql_order_line",
		"bmsql_stock",
		"bmsql_warehouse",
	}
	adminSQL := "admin check table "
	for _, v := range tableNames {
		if _, err := tidbServer.Exec(adminSQL + v); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

var consistencyCheckSQLs = []string{
	`-- SET search_path TO TPCC;
-- Condition 1: W_YTD = sum(D_YTD)
SELECT * FROM (SELECT w.w_id, w.w_ytd, d.sum_d_ytd
FROM bmsql_warehouse w,
(SELECT d_w_id, SUM(d_ytd) sum_d_ytd
FROM bmsql_district
GROUP BY d_w_id) d
WHERE w.w_id = d.d_w_id) as x
WHERE w_ytd != sum_d_ytd;`,

	`-- Condition 2: D_NEXT_O_ID - 1 = max(O_ID) = max(NO_O_ID)
SELECT * FROM (SELECT d.d_w_id, d.d_id, d.d_next_o_id, o.max_o_id, no.max_no_o_id
FROM bmsql_district d,
(SELECT o_w_id, o_d_id, MAX(o_id) max_o_id
FROM bmsql_oorder
GROUP BY o_w_id, o_d_id) o,
(SELECT no_w_id, no_d_id, MAX(no_o_id) max_no_o_id
FROM bmsql_new_order
GROUP BY no_w_id, no_d_id) no
WHERE d.d_w_id = o.o_w_id AND d.d_w_id = no.no_w_id AND
d.d_id = o.o_d_id AND d.d_id = no.no_d_id) as x
WHERE d_next_o_id - 1 != max_o_id OR d_next_o_id - 1 != max_no_o_id;`,

	`-- Condition 3: max(NO_O_ID) - min(NO_O_ID) + 1
-- = [number of rows in the NEW-ORDER table for this bmsql_district]
SELECT * FROM (SELECT no_w_id, no_d_id, MAX(no_o_id) max_no_o_id,
MIN(no_o_id) min_no_o_id, COUNT(*) count_no
FROM bmsql_new_order
GROUP BY no_w_id, no_d_Id) as x
WHERE max_no_o_id - min_no_o_id + 1 != count_no;`,

	`-- Condition 4: sum(O_OL_CNT)
-- = [number of rows in the ORDER-LINE table for this bmsql_district]
SELECT * FROM (SELECT o.o_w_id, o.o_d_id, o.sum_o_ol_cnt, ol.count_ol
FROM (SELECT o_w_id, o_d_id, SUM(o_ol_cnt) sum_o_ol_cnt
FROM bmsql_oorder
GROUP BY o_w_id, o_d_id) o,
(SELECT ol_w_id, ol_d_id, COUNT(*) count_ol
FROM bmsql_order_line
GROUP BY ol_w_id, ol_d_id) ol
WHERE o.o_w_id = ol.ol_w_id AND
o.o_d_id = ol.ol_d_id) as x
WHERE sum_o_ol_cnt != count_ol;`,

	`-- Condition 5: For any row in the ORDER table, O_CARRIER_ID is set to a null
-- value if and only if there is a corresponding row in the
-- NEW-ORDER table
SELECT * FROM (SELECT o.o_w_id, o.o_d_id, o.o_id, o.o_carrier_id, no.count_no
FROM bmsql_oorder o,
(SELECT no_w_id, no_d_id, no_o_id, COUNT(*) count_no
FROM bmsql_new_order
GROUP BY no_w_id, no_d_id, no_o_id) no
WHERE o.o_w_id = no.no_w_id AND
o.o_d_id = no.no_d_id AND
o.o_id = no.no_o_id) as x
WHERE (o_carrier_id IS NULL AND count_no = 0) OR
(o_carrier_id IS NOT NULL AND count_no != 0);`,

	`-- Condition 6: For any row in the ORDER table, O_OL_CNT must equal the number
-- of rows in the ORDER-LINE table for the corresponding order
SELECT * FROM (SELECT o.o_w_id, o.o_d_id, o.o_id, o.o_ol_cnt, ol.count_ol
FROM bmsql_oorder o,
(SELECT ol_w_id, ol_d_id, ol_o_id, COUNT(*) count_ol
FROM bmsql_order_line
GROUP BY ol_w_id, ol_d_id, ol_o_id) ol
WHERE o.o_w_id = ol.ol_w_id AND
o.o_d_id = ol.ol_d_id AND
o.o_id = ol.ol_o_id) as x
WHERE o_ol_cnt != count_ol;`,

	`-- Condition 7: For any row in the ORDER-LINE table, OL_DELIVERY_D is set to
-- a null date/time if and only if the corresponding row in the
-- ORDER table has O_CARRIER_ID set to a null value
SELECT * FROM (SELECT ol.ol_w_id, ol.ol_d_id, ol.ol_o_id, ol.ol_delivery_d,
o.o_carrier_id
FROM bmsql_order_line ol,
bmsql_oorder o
WHERE ol.ol_w_id = o.o_w_id AND
ol.ol_d_id = o.o_d_id AND
ol.ol_o_id = o.o_id) as x
WHERE (ol_delivery_d IS NULL AND o_carrier_id IS NOT NULL) OR
(ol_delivery_d IS NOT NULL AND o_carrier_id IS NULL);`,

	`-- Condition 8: W_YTD = sum(H_AMOUNT)
SELECT * FROM (SELECT w.w_id, w.w_ytd, h.sum_h_amount
FROM bmsql_warehouse w,
(SELECT h_w_id, SUM(h_amount) sum_h_amount
FROM bmsql_history
GROUP BY h_w_id) h
WHERE w.w_id = h.h_w_id) as x
WHERE w_ytd != sum_h_amount;`,

	`-- Condition 9: D_YTD = sum(H_AMOUNT)
SELECT * FROM (SELECT d.d_w_id, d.d_id, d.d_ytd, h.sum_h_amount
FROM bmsql_district d,
(SELECT h_w_id, h_d_id, SUM(h_amount) sum_h_amount
FROM bmsql_history
GROUP BY h_w_id, h_d_id) h
WHERE d.d_w_id = h.h_w_id AND
d.d_id = h.h_d_id) as x
WHERE d_ytd != sum_h_amount;`,
}