// Copyright 2019 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package txn

import (
	"context"
	"database/sql"
	"fmt"

	"github.com/pingcap/errors"
	"github.com/pingcap/schrodinger-test/randsql"
	client "github.com/pingcap/schrodinger/pkg/test_client"
)

func (s *testSuit) checkRandSelect(ctx context.Context, tiConn, myConn *sql.Conn) error {
	task := s.rg.GenRandSelect()
	if task == nil {
		return nil
	}
	return s.checkSelect(ctx, tiConn, myConn, task)
}

func (s *testSuit) checkSelectWithTableName(ctx context.Context, tiConn, myConn *sql.Conn, tableName string) error {
	task := s.rg.GenRandSelectWithTableName(tableName)
	if task == nil {
		return nil
	}
	return s.checkSelect(ctx, tiConn, myConn, task)
}

func (s *testSuit) checkSelect(ctx context.Context, tiConn, myConn *sql.Conn, task *randsql.DMLJobTask) error {
	tiRows, tiErr := tiConn.QueryContext(ctx, task.SQL())
	myRows, myErr := myConn.QueryContext(ctx, task.SQL())
	if tiRows != nil {
		defer tiRows.Close()
	}
	if myRows != nil {
		defer myRows.Close()
	}
	client.Infof("SQL: %v, tidb: %v, mysql: %v", task.SQL(), tiErr, myErr)
	if tiErr != nil || myErr != nil {
		return fmt.Errorf("Error when executing SQL: %s\n remote tidb Err: %#v\n remote mysql err: %#v\n%v\n", task.SQL(), tiErr, myErr, task.Table().DebugPrintToString())
	}
	tiRowsMap, tiErr := getDataFromDBRows(tiRows)
	myRowsMap, myErr := getDataFromDBRows(myRows)
	if tiErr != nil || myErr != nil {
		return fmt.Errorf("Error when query SQL: %s\n remote tidb Err: %#v\n remote mysql err: %#v\n%v\n", task.SQL(), tiErr, myErr, task.Table().DebugPrintToString())
	}
	if err := checkEqualRows(tiRowsMap, myRowsMap); err != nil {
		return errors.Annotatef(err, "Error when executing SQL: %s\n tidb rows: %#v\n mysql rows: %#v\n%v\n", task.SQL(), tiRowsMap, myRowsMap, task.Table().DebugPrintToString())
	}
	return nil
}

func (s *testSuit) adminCheckTable() error {
	task := s.rg.GenAdminCheck()
	if task == nil {
		return nil
	}
	_, err := s.tiDB.Exec(task.SQL())
	client.Infof("SQL: %v, tidb: %v", task.SQL(), err)
	if err != nil {
		if randsql.IgnoreDMLError(err) {
			return nil
		}
		return errors.Annotatef(err, "Error when executing SQL: %s", task.SQL())
	}
	return nil
}

func checkEqualRows(tiRows, myRows map[string]int) error {
	if len(tiRows) != len(myRows) {
		return errors.Errorf("tidb rows len %v != mysql rows len: %v", len(tiRows), len(myRows))
	}
	for k, v1 := range tiRows {
		v2, ok := myRows[k]
		if !ok || v1 != v2 {
			return errors.Errorf("row %v, tidb %v != mysql %v", k, v1, v2)
		}
	}
	return nil
}

func getDataFromDBRows(rows *sql.Rows) (map[string]int, error) {
	actualRows, err := randsql.GetDataFromDBRows(rows)
	if err != nil {
		return nil, err
	}
	actualRowsMap := make(map[string]int)
	for _, row := range actualRows {
		rowString := ""
		for _, col := range row {
			rowString += fmt.Sprintf("%v,", col)
		}
		_, ok := actualRowsMap[rowString]
		if !ok {
			actualRowsMap[rowString] = 0
		}
		actualRowsMap[rowString]++
	}
	return actualRowsMap, nil
}

func mustExec(ctx context.Context, conn *sql.Conn, sql string) {
	_, err := conn.ExecContext(ctx, sql)
	assertNil(err)
}

func dbMustExec(db *sql.DB, sql string) {
	_, err := db.Exec(sql)
	assertNil(err)
}

func (s *testSuit) mustCreateTable() {
	task := s.rg.GenRandCreateTable()
	ctx := context.Background()
	tiConn, err := s.tiDB.Conn(ctx)
	assertNil(err)
	defer tiConn.Close()
	myConn, err := s.myDB.Conn(ctx)
	assertNil(err)
	defer myConn.Close()
	err = s.execSQL(ctx, task, tiConn, myConn)
	assertNil(err)
}

func assertNil(err error) {
	if err == nil {
		return
	}
	client.Fatalf("assert error nil failed. err: %v\n", err)
}