package storage

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/jmoiron/sqlx"
	stypes "github.com/jmoiron/sqlx/types"
	"github.com/juju/errors"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/schrodinger/util"
)

// CatProcessConditions conducts the storage query
type CatProcessConditions struct {
	Stage     []string
	Status    []string
	Destroyed []string
	Names     []string
	Limit     int64
	Offset    int64
}

/* ---------------- mysql storage ------------------*/

func (m *mysqlClient) createCatProcess(cat *types.CatProcess) error {
	t := time.Now().Format(util.TimeFormat)
	cat.Ctime, cat.Utime = t, t

	txn, err := m.db.Beginx()
	if err != nil {
		return errors.Trace(err)
	}

	res, err := m.db.NamedExec(stmtCatProcessInsert, catProcessToParams(cat))
	if err != nil {
		txn.Rollback()
		return errors.Trace(err)
	}

	cat.ID, err = res.LastInsertId()
	if err != nil {
		txn.Rollback()
		return errors.Trace(err)
	}

	for _, versionInfo := range []*types.VersionInfo{&cat.RunVersion.TiKVVersion, &cat.RunVersion.PDVersion, &cat.RunVersion.TiDBVersion} {
		if util.IsEmpty(versionInfo.BuildTime) {
			continue
		}
		versionInfo.CatProcessID = cat.ID
		_, err = m.db.NamedExec(stmtComponentVersionInsert, componentVersionToParams(versionInfo))
		if err != nil {
			txn.Rollback()
			return errors.Trace(err)
		}
	}

	if err = txn.Commit(); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func catProcessToParams(cat *types.CatProcess) map[string]interface{} {
	return map[string]interface{}{
		"id":              cat.ID,
		"labels":          encodeJSONParams(cat.Meta.Labels),
		"box":             cat.BoxID,
		"box_name":        cat.BoxName,
		"experiment_name": cat.ExperimentName,
		// TODO: set cat id
		"cat":           0,
		"owner":         cat.Owner,
		"creator":       cat.Creator,
		"create_time":   cat.Ctime,
		"update_time":   cat.Utime,
		"name":          cat.Name,
		"namespace":     cat.Namespace,
		"destroyed":     cat.Destroyed,
		"config":        encodeJSONParams(cat.Config),
		"state":         cat.State,
		"stage":         cat.Stage,
		"report":        encodeJSONParams(cat.Report),
		"use_box_label": cat.UseBoxLabel,
		"ignore_labels": encodeJSONParams(cat.IgnoreLabels),
		"backup_name":   cat.BackupName,
	}
}

func componentVersionToParams(cv *types.VersionInfo) map[string]interface{} {
	return map[string]interface{}{
		"name":           cv.Name,
		"type":           cv.Type,
		"value":          cv.Value,
		"platform":       cv.Platform,
		"compile_option": cv.CompileOption,
		"build_time":     cv.BuildTime,
		"cat_process":    cv.CatProcessID,
	}
}

func (m *mysqlClient) updateCatProcess(cat *types.CatProcess) error {
	cat.Utime = time.Now().Format(util.TimeFormat)

	txn, err := m.db.Beginx()
	if err != nil {
		return errors.Trace(err)
	}

	_, err = m.db.NamedExec(stmtComponentVersionDelete, map[string]interface{}{"cat_process": cat.ID})
	if err != nil {
		txn.Rollback()
		return errors.Trace(err)
	}

	_, err = m.db.NamedExec(stmtCatProcessUpdate, catProcessToParams(cat))
	if err != nil {
		txn.Rollback()
		return errors.Trace(err)
	}

	for _, versionInfo := range []*types.VersionInfo{&cat.RunVersion.TiKVVersion, &cat.RunVersion.PDVersion, &cat.RunVersion.TiDBVersion} {
		if util.IsEmpty(versionInfo.BuildTime) {
			continue
		}
		versionInfo.CatProcessID = cat.ID
		_, err = m.db.NamedExec(stmtComponentVersionInsert, componentVersionToParams(versionInfo))
		if err != nil {
			txn.Rollback()
			return errors.Trace(err)
		}
	}

	if err = txn.Commit(); err != nil {
		txn.Rollback()
		return errors.Trace(err)
	}

	return nil
}

func (m *mysqlClient) findCatProcess(id int64) (*types.CatProcess, error) {
	txn, err := m.db.Beginx()
	if err != nil {
		return nil, errors.Trace(err)
	}

	query, args, err := m.db.BindNamed(stmtCatProcessQuery, map[string]interface{}{"id": id})
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	row := m.db.QueryRowx(query, args...)
	if row == nil {
		txn.Rollback()
		return nil, errors.NotFoundf("cat process %d", id)
	}

	out := &types.CatProcess{
		Report:       &types.CatReport{},
		Meta:         &types.Meta{Labels: make(map[string]string)},
		IgnoreLabels: make(map[string]string),
	}

	if err = scanCatProcessRow(row, out); err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	if err = m.setComponentVersion(out); err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	if err = txn.Commit(); err != nil {
		return nil, errors.Trace(err)
	}

	return out, nil
}

func (m *mysqlClient) findCatProcessByName(name string) (*types.CatProcess, error) {
	txn, err := m.db.Beginx()
	if err != nil {
		return nil, errors.Trace(err)
	}

	query, args, err := m.db.BindNamed(stmtCatProcessQueryByName, map[string]interface{}{"name": name})
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	row := m.db.QueryRowx(query, args...)
	if row == nil {
		txn.Rollback()
		return nil, errors.NotFoundf("cat process %s", name)
	}

	out := &types.CatProcess{
		Report:       &types.CatReport{},
		Meta:         &types.Meta{Labels: make(map[string]string)},
		IgnoreLabels: make(map[string]string),
	}

	if err := scanCatProcessRow(row, out); err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	if err := m.setComponentVersion(out); err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	if err = txn.Commit(); err != nil {
		return nil, errors.Trace(err)
	}

	return out, nil
}

func (m *mysqlClient) findCatProcessByExperiments(names []string) ([]*types.CatProcess, error) {
	txn, err := m.db.Beginx()
	if err != nil {
		return nil, errors.Trace(err)
	}

	var values []string
	for _, name := range names {
		values = append(values, fmt.Sprintf("'%s'", name))
	}
	experimentNames := strings.Join(values, ",")
	stmtCatProcessQueryByExperiments := fmt.Sprintf("%s WHERE experiment_name IN (%s) ORDER BY create_time DESC", stmtCatProcessQueryBase, experimentNames)

	rows, err := txn.Queryx(stmtCatProcessQueryByExperiments)
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	cps, err := scanCatProcessRows(rows)
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	var cpIDs []int64
	ID2cp := make(map[int64]*types.CatProcess)

	for _, cp := range cps {
		cpIDs = append(cpIDs, cp.ID)
		ID2cp[cp.ID] = cp
	}

	cvs, err := m.findComponentVersionByCatProcesses(cpIDs)
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	for _, cv := range cvs {
		cpID := cv.CatProcessID
		if cp, ok := ID2cp[cpID]; ok {
			switch cv.Name {
			case "pd":
				cp.RunVersion.PDVersion = *cv
			case "tidb":
				cp.RunVersion.TiDBVersion = *cv
			case "tikv":
				cp.RunVersion.TiKVVersion = *cv
			}
		}
	}

	if err = txn.Commit(); err != nil {
		return nil, errors.Trace(err)
	}

	return cps, nil
}

func (m *mysqlClient) findCatProcessByExperiment(name string) ([]*types.CatProcess, error) {
	txn, err := m.db.Beginx()
	if err != nil {
		return nil, errors.Trace(err)
	}

	query, args, err := m.db.BindNamed(stmtCatProcessQueryByExperiment, map[string]interface{}{"experiment_name": name})
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	rows, err := m.db.Queryx(query, args...)
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	cps, err := scanCatProcessRows(rows)
	if err != nil {
		txn.Rollback()
		return nil, errors.Trace(err)
	}

	for _, cp := range cps {
		if err := m.setComponentVersion(cp); err != nil {
			txn.Rollback()
			return nil, errors.Trace(err)
		}
	}

	if err = txn.Commit(); err != nil {
		return nil, errors.Trace(err)
	}

	return cps, nil
}

func (m *mysqlClient) setComponentVersion(cp *types.CatProcess) error {
	cvs, err := m.findComponentVersionByCatProcess(cp.ID)
	if err != nil {
		return errors.Trace(err)
	}

	// TODO: need to refine
	for _, cv := range cvs {
		switch cv.Name {
		case "pd":
			cp.RunVersion.PDVersion = *cv
		case "tidb":
			cp.RunVersion.TiDBVersion = *cv
		case "tikv":
			cp.RunVersion.TiKVVersion = *cv
		}
	}

	return nil
}

func (m *mysqlClient) listCatProcess(conditions *CatProcessConditions) ([]*types.CatProcess, int64, error) {
	tx, err := m.db.Beginx()

	var total int64
	if err := tx.Get(&total, genCatProcessCountStmt(conditions)); err != nil {
		tx.Rollback()
		return nil, 0, errors.Trace(err)
	}

	rows, err := tx.Queryx(genCatProcessListStmt(conditions))
	if err != nil {
		tx.Rollback()
		return nil, 0, errors.Trace(err)
	}

	cps, err := scanCatProcessRows(rows)
	if err != nil {
		tx.Rollback()
		return nil, 0, errors.Trace(err)
	}

	for _, cp := range cps {
		if err := m.setComponentVersion(cp); err != nil {
			tx.Rollback()
			return nil, 0, errors.Trace(err)
		}
	}

	if err := tx.Commit(); err != nil {
		return nil, 0, errors.Trace(err)
	}

	return cps, total, nil
}

func scanCatProcessRows(rows *sqlx.Rows) ([]*types.CatProcess, error) {
	defer rows.Close()

	cps := []*types.CatProcess{}
	for rows.Next() {
		cp := &types.CatProcess{
			Report:       &types.CatReport{},
			Meta:         &types.Meta{Labels: make(map[string]string)},
			IgnoreLabels: make(map[string]string),
		}

		if err := scanCatProcessRow(rows, cp); err != nil {
			return nil, errors.Trace(err)
		}

		cps = append(cps, cp)
	}

	return cps, nil
}

func scanCatProcessRow(scanner Scanner, dest *types.CatProcess) error {
	labelsJSON := stypes.JSONText{}
	ignoreLabelsJSON := stypes.JSONText{}
	reportJSON := stypes.JSONText{}
	configJSON := stypes.JSONText{}
	backupName := sql.NullString{}
	err := scanner.Scan(
		&dest.ID,
		&labelsJSON,
		&dest.BoxID,
		&dest.BoxName,
		&dest.ExperimentName,
		&dest.Owner,
		&dest.Creator,
		&dest.Ctime,
		&dest.Utime,
		&dest.Name,
		&dest.Namespace,
		&dest.Destroyed,
		&configJSON,
		&dest.State,
		&dest.Stage,
		&reportJSON,
		&dest.UseBoxLabel,
		&ignoreLabelsJSON,
		&backupName,
	)
	json.Unmarshal(labelsJSON, &dest.Meta.Labels)
	json.Unmarshal(ignoreLabelsJSON, &dest.IgnoreLabels)
	json.Unmarshal(reportJSON, &dest.Report)
	json.Unmarshal(configJSON, &dest.Config)

	dest.BackupName = backupName.String

	return errors.Trace(err)
}

func (m *mysqlClient) findComponentVersionByCatProcess(catProcessID int64) ([]*types.VersionInfo, error) {
	params := map[string]interface{}{
		"cat_process": catProcessID,
	}

	rows, err := m.db.NamedQuery(stmtComponentVersionQueryByCatProcess, params)
	if err != nil {
		return nil, errors.Trace(err)
	}

	cvs, err := scanComponentVersionRows(rows)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return cvs, nil
}

func (m *mysqlClient) findComponentVersionByCatProcesses(catProcessIDs []int64) ([]*types.VersionInfo, error) {
	var catIDs []string
	for _, catID := range catProcessIDs {
		catIDs = append(catIDs, fmt.Sprintf("%d", catID))
	}
	var stmtComponentVersionQueryByCatProcesses = stmtComponentVersionQueryBase + " WHERE 1 = 1" + genInClause("cat_process", catIDs)

	rows, err := m.db.Queryx(stmtComponentVersionQueryByCatProcesses)
	if err != nil {
		return nil, errors.Trace(err)
	}

	cvs, err := scanComponentVersionRows(rows)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return cvs, nil
}

func scanComponentVersionRows(rows *sqlx.Rows) ([]*types.VersionInfo, error) {
	defer rows.Close()

	var cvs []*types.VersionInfo
	for rows.Next() {
		cv := &types.VersionInfo{}
		if err := scanComponentVersionRow(rows, cv); err != nil {
			return nil, errors.Trace(err)
		}

		cvs = append(cvs, cv)
	}

	return cvs, nil
}

func scanComponentVersionRow(scanner Scanner, dest *types.VersionInfo) error {
	err := scanner.Scan(
		&dest.Name,
		&dest.Type,
		&dest.Value,
		&dest.Platform,
		&dest.CompileOption,
		&dest.BuildTime,
		&dest.CatProcessID,
	)

	return errors.Trace(err)
}

const stmtCatProcessInsert = `
INSERT INTO cat_process (
 labels
,box
,box_name
,experiment_name
,cat
,owner
,creator
,create_time
,update_time
,name
,namespace
,destroyed
,config
,state
,stage
,report
,use_box_label
,ignore_labels
,backup_name
) VALUES (
 :labels
,:box
,:box_name
,:experiment_name
,:cat
,:owner
,:creator
,:create_time
,:update_time
,:name
,:namespace
,:destroyed
,:config
,:state
,:stage
,:report
,:use_box_label
,:ignore_labels
,:backup_name
)
`
const stmtCatProcessUpdate = `
UPDATE cat_process SET
 labels = :labels
,box = :box
,box_name = :box_name
,experiment_name = :experiment_name
,cat = :cat
,owner = :owner
,creator = :creator
,update_time = :update_time
,name = :name
,namespace = :namespace
,destroyed = :destroyed
,config = :config
,state = :state
,stage = :stage
,report = :report
,use_box_label = :use_box_label
,ignore_labels = :ignore_labels
,backup_name = :backup_name
WHERE id = :id
`

const stmtComponentVersionInsert = `
INSERT INTO component_version (
 name
,type
,value
,platform
,compile_option
,build_time
,cat_process
) VALUES (
 :name
, :type
,:value
,:platform
,:compile_option
,:build_time
,:cat_process
)
`

const stmtComponentVersionDelete = `
DELETE FROM component_version
WHERE cat_process = :cat_process
`

const stmtCatProcessQueryBase = `
SELECT
 id
,labels
,box
,box_name
,experiment_name
,owner
,creator
,create_time
,update_time
,name
,namespace
,destroyed
,config
,state
,stage
,report
,use_box_label
,ignore_labels
,backup_name
FROM cat_process
`

const stmtCatProcessCount = `
SELECT COUNT(*)
FROM cat_process
`

const stmtCatProcessQuery = stmtCatProcessQueryBase + `
WHERE id = :id
`

const stmtCatProcessQueryByName = stmtCatProcessQueryBase + `
WHERE name = :name
`

const stmtCatProcessQueryByExperiment = stmtCatProcessQueryBase + `
WHERE experiment_name = :experiment_name
ORDER BY create_time DESC
`

func genCatProcessListStmt(conditions *CatProcessConditions) string {
	var catProcessSelectTemplate = stmtCatProcessQueryBase + `WHERE %s
ORDER BY create_time DESC
`
	var conds = "1 = 1"

	if conditions != nil {
		conds = conds + genLikeClause("name", conditions.Names)
		conds = conds + genInClause("stage", conditions.Stage)
		conds = conds + genInClause("state", conditions.Status)
		conds = conds + genInClause("destroyed", conditions.Destroyed)
	}

	result := fmt.Sprintf(catProcessSelectTemplate, conds)
	if conditions != nil && conditions.Limit != 0 {
		result = result + fmt.Sprintf("LIMIT %d OFFSET %d\n", conditions.Limit, conditions.Offset)
	}
	return result
}

func genCatProcessCountStmt(conditions *CatProcessConditions) string {
	var catProcessCountTemplate = stmtCatProcessCount + `WHERE %s
ORDER BY create_time DESC
`
	var conds = "1 = 1"

	if conditions != nil {
		conds = conds + genLikeClause("name", conditions.Names)
		conds = conds + genInClause("stage", conditions.Stage)
		conds = conds + genInClause("state", conditions.Status)
		conds = conds + genInClause("destroyed", conditions.Destroyed)
	}

	result := fmt.Sprintf(catProcessCountTemplate, conds)
	return result
}

const stmtComponentVersionQueryBase = `
SELECT
 name
,type
,value
,platform
,compile_option
,build_time
,cat_process
FROM component_version
`

const stmtComponentVersionQueryByCatProcess = stmtComponentVersionQueryBase + `
WHERE cat_process = :cat_process
`