// Copyright 2017 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 main

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/juju/errors"
	"github.com/pingcap/schrodinger-test/util"
	client "github.com/pingcap/schrodinger/pkg/test_client"
	"github.com/pingcap/schrodinger/pkg/types"
	"golang.org/x/net/context"
)

const (
	mysqlOutDir = "mysql_r"
	tidbOutDir  = "tidb_r"
	costOutDir  = "cost_r"
)

type TPCHConfig struct {
	Host       string `toml:"host"`
	Port       int    `toml:"port"`
	ScriptsDir string `toml:"scripts_dir"`
}

type TPCHSuite struct {
	cfg *TPCHConfig
}

func NewTPCHSuite(cfg *TPCHConfig) *TPCHSuite {
	return &TPCHSuite{cfg: cfg}
}

func (s *TPCHSuite) name() string {
	return "tpch"
}

func getFileScanner(fileName string) (*bufio.Scanner, error) {
	file, err := os.Open(fileName)
	if err != nil {
		return nil, err
	}

	return bufio.NewScanner(file), nil
}

// DiffType defines the type of a diff item.
type DiffType int8

const (
	diffNone DiffType = iota
	diffFileLength
	diffLineContent
	fileEqual
)

func diffTypeToString(tp DiffType) string {
	switch tp {
	case diffNone:
		return "diffNone"
	case diffFileLength:
		return "diffFileLength"
	case diffLineContent:
		return "diffLineContent"
	case fileEqual:
		return "fileEqual"
	default:
		return "Unknown"
	}
}

type lineDiff struct {
	line   int
	column int
}

func (s *TPCHSuite) diffFiles(f1, f2 string) (tp DiffType, diffs []lineDiff, err error) {
	scanner1, err := getFileScanner(f1)
	if err != nil {
		return diffNone, nil, errors.Trace(err)
	}
	scanner2, err := getFileScanner(f2)
	if err != nil {
		return diffNone, nil, errors.Trace(err)
	}

	f1Eof, f2Eof := false, false
	li, col := 0, 0
	for {
		if !scanner1.Scan() {
			f1Eof = true
		}

		if !scanner2.Scan() {
			f2Eof = true
		}

		if f1Eof || f2Eof {
			break
		}

		line1 := scanner1.Text()
		line2 := scanner2.Text()
		seps1 := strings.Split(line1, "\t")
		seps2 := strings.Split(line2, "\t")

		if len(seps1) != len(seps2) {
			diff := lineDiff{line: li, column: col}
			diffs = append(diffs, diff)
			continue
		}

		for i := 0; i < len(seps1); i++ {
			sep1 := seps1[i]
			sep2 := seps2[i]
			if sep1 != sep2 {
				f1, err := strconv.ParseFloat(sep1, 64)
				if err != nil {
					diff := lineDiff{line: li, column: col}
					diffs = append(diffs, diff)
					// only report the first diff in one line
					break
				}
				f2, err := strconv.ParseFloat(sep2, 64)
				if err != nil {
					diff := lineDiff{line: li, column: col}
					diffs = append(diffs, diff)
					// only report the first diff in one line
					break
				}
				n1 := fmt.Sprintf("%.8f", f1)
				n2 := fmt.Sprintf("%.8f", f2)
				if n1 != n2 {
					diff := lineDiff{line: li, column: col}
					diffs = append(diffs, diff)
					// only report the first diff in one line
					break
				}
			}
			col++
		}
		li++
		col = 0
	}

	if err = scanner1.Err(); err != nil {
		return diffNone, nil, errors.Trace(err)
	}

	if err = scanner2.Err(); err != nil {
		return diffNone, nil, errors.Trace(err)
	}

	// file length is not equal
	if f1Eof == false || f2Eof == false {
		return diffFileLength, diffs, nil
	}

	if len(diffs) > 0 {
		return diffLineContent, diffs, nil
	}
	return fileEqual, diffs, nil
}

func (s *TPCHSuite) resultFile(resultName string, index int) string {
	return fmt.Sprintf("%s/%s/q%d.out", s.cfg.ScriptsDir, resultName, index)
}

func (s *TPCHSuite) checkAnswers() (*types.TestResult, error) {
	caseRs := &types.TestResult{
		TestName: s.name(),
	}
	for i := 1; i < 23; i++ {
		if i == 15 {
			// skip view test
			continue
		}
		mysqlOut := s.resultFile(mysqlOutDir, i)
		tidbOut := s.resultFile(tidbOutDir, i)
		costOut := s.costFile(costOutDir, i)

		cost, err := s.fetchCost(costOut)
		if err != nil {
			return nil, errors.Trace(err)
		}

		tp, diffs, err := s.diffFiles(mysqlOut, tidbOut)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if tp != fileEqual {
			client.Errorf("[tpch failed]: Query:%d %v", i, diffTypeToString(tp))
			for _, diff := range diffs {
				client.Errorf("Query:%d Result Unmatch (Line, Column):(%v, %v)\n", i, diff.line, diff.column)
			}
		}

		caseRs.QueryResults = append(caseRs.QueryResults, &types.QueryResult{
			Query: fmt.Sprintf("sql%d", i),
			Cost:  cost,
		})
	}
	return caseRs, nil
}

func (s *TPCHSuite) fetchCost(fileName string) (time.Duration, error) {
	file, err := os.Open(fileName)
	if err != nil {
		return 0, err
	}

	readder := bufio.NewReader(file)
	line, _, err := readder.ReadLine()
	if err != nil && err != io.EOF {
		return 0, err
	}

	cost, err := strconv.ParseInt(strings.TrimSpace(string(line)), 10, 64)
	if err != nil {
		return 0, err
	}

	return time.Duration(cost) * time.Second, nil
}

func (s *TPCHSuite) costFile(resultName string, index int) string {
	return fmt.Sprintf("%s/%s/q%d.time", s.cfg.ScriptsDir, resultName, index)
}

func (s *TPCHSuite) run(ctx context.Context) (*types.TestResult, error) {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	cmd := fmt.Sprintf("/bin/sh %s/run-tpch.sh tidb_r %s %d", s.cfg.ScriptsDir, s.cfg.Host, s.cfg.Port)

	client.Infof("[tpch]start to exec cmd: %s", cmd)
	output, err := util.ExecCmd(ctx, cmd)
	if err != nil {
		client.Errorf("[tpch] exec [%s] failed: %v, out: %v", cmd, err, output)
		return nil, errors.Trace(err)
	}

	return s.checkAnswers()
}

func (s *TPCHSuite) loadData(ctx context.Context) error {
	cctx, _ := context.WithCancel(ctx)
	dbDSN := fmt.Sprintf("root:@tcp(%s:%d)/test", s.cfg.Host, s.cfg.Port)
	db, err := util.OpenDB(dbDSN, 1)
	if err != nil {
		client.Errorf("[tpch] create db client error %v", err)
		return errors.Trace(err)
	}
	util.MustExec(db, tableSchema)

	cmd := fmt.Sprintf("/bin/sh %s/loadalldata.sh %s %d", s.cfg.ScriptsDir, s.cfg.Host, s.cfg.Port)
	client.Infof("[tpch] start to exec cmd: %s", cmd)
	output, err := util.ExecCmd(cctx, cmd)
	if err != nil {
		client.Errorf("[tpch] exec [%s] failed: %v, out: %v", cmd, err, output)
		return errors.Trace(err)
	}
	client.Debugf("[tpch] exec [%s] success, output: %s", cmd, output)

	return nil
}