// 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 testcases

import (
	"strconv"

	"github.com/pingcap/test-infra/caselib/pkg/host"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"github.com/pingcap/test-infra/caselib/pkg/steps"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	"github.com/pingcap/test-infra/common/model/resource"
)

const (
	ycsbTable = "usertable"
)

// CommonEnv init from common used testbed
// common used testbed contains: x upstream, 1 downstream cluster, 1 dm cluster, 1 workload, 1 for sync diff,
//   0 or 1 osc instance
// resource item names should follow the naming rule: see init below
type CommonEnv struct {
	b *steps.Builder

	withOsc  bool
	withTiup bool

	UpStreams  []*resource.ResourceHandle
	DownStream *resource.ResourceHandle
	Dm         *resource.ResourceHandle
	Workload   *resource.ResourceHandle
	SyncDiff   *resource.ResourceHandle
	Osc        *resource.ResourceHandle
	Tiup       *resource.ResourceHandle
	DmHost     *host.DMHost
}

func newCommonEnv() *CommonEnv {
	return &CommonEnv{
		b:          steps.NewBuilder(),
		UpStreams:  []*resource.ResourceHandle{},
		DownStream: &resource.ResourceHandle{},
		Dm:         &resource.ResourceHandle{},
		Workload:   &resource.ResourceHandle{},
		SyncDiff:   &resource.ResourceHandle{},
		Osc:        &resource.ResourceHandle{},
		Tiup:       &resource.ResourceHandle{},
		DmHost:     &host.DMHost{},
	}
}

func newCommonEnvWithOsc() *CommonEnv {
	env := newCommonEnv()
	env.withOsc = true
	return env
}

func newCommonEnvWithTiup() *CommonEnv {
	env := newCommonEnv()
	env.withTiup = true
	return env
}

// env init
// 1. load resource
// 2. init dm host
// 3. adjust comfig by replacing real host
func (env *CommonEnv) init() *CommonEnv {
	env.b.LoadEnv()

	for i := 0; i < len(Path.SourcePaths); i++ {
		handle := &resource.ResourceHandle{}
		env.b.LoadResourceFromEnv(utils.ResourcePrefix+"upstream"+strconv.Itoa(i+1), handle)
		env.UpStreams = append(env.UpStreams, handle)
	}

	env.b.LoadResourceFromEnv(utils.ResourcePrefix+"downstream", env.DownStream)
	env.b.LoadResourceFromEnv(utils.ResourcePrefix+"dm", env.Dm)
	env.b.LoadResourceFromEnv(utils.ResourcePrefix+"workload", env.Workload)
	env.b.LoadResourceFromEnv(utils.ResourcePrefix+"syncdiff", env.SyncDiff)
	if env.withOsc {
		env.b.LoadResourceFromEnv(utils.ResourcePrefix+"osc", env.Osc)
	}
	if env.withTiup {
		env.b.LoadResourceFromEnv(utils.ResourcePrefix+"tiup", env.Tiup)
	}
	env.b.InitDMHost(env.Dm, env.DownStream, env.SyncDiff, env.UpStreams, env.DmHost)
	env.b.AdjustConfig(env.Dm, env.SyncDiff, env.DmHost, Path)
	return env
}

func (env *CommonEnv) initDbsAndSourcesAndTasks(dbName string) *CommonEnv {
	for i := 0; i < len(env.UpStreams); i++ {
		env.b.CreateDatabase(env.UpStreams[i], dbName)
	}
	for _, sourcePath := range Path.SourcePaths {
		env.b.DmCreateSource(env.Dm, env.DmHost, sourcePath)
	}
	for _, taskPath := range Path.TaskPaths {
		env.b.DmStartTask(env.Dm, env.DmHost, taskPath)
	}
	return env
}

func (env *CommonEnv) initDbsAndSourcesAndTasksWithPrepareData(dbName string) *CommonEnv {
	for i := 0; i < len(env.UpStreams); i++ {
		env.b.CreateDatabase(env.UpStreams[i], dbName)
	}
	env.b.DmRunWorkload(env.UpStreams[0], env.Dm, env.Workload, Spec.GetWorkloadOption(workload.CommandTypePrepare))
	for _, sourcePath := range Path.SourcePaths {
		env.b.DmCreateSource(env.Dm, env.DmHost, sourcePath)
	}
	for _, taskPath := range Path.TaskPaths {
		env.b.DmStartTask(env.Dm, env.DmHost, taskPath)
	}
	return env
}

// runBasic is same as dm_migrate_basic_xxx.
func (env *CommonEnv) runBasic(dbName string, taskName string, sourceNameList []string) *CommonEnv {
	env.b.CreateDatabase(env.UpStreams[0], dbName).
		DmRunWorkload(env.UpStreams[0], env.Dm, env.Workload, Spec.GetWorkloadOption(workload.CommandTypePrepare)).
		DmCreateSources(env.Dm, env.DmHost, Path.SourcePaths).
		DmStartTasks(env.Dm, env.DmHost, Path.TaskPaths).
		DmWaitTaskSynced(env.DmHost, taskName, sourceNameList).
		DmCheckSyncDiff(env.Dm, env.SyncDiff, env.DmHost, Path.DiffPaths[0]).
		DmRunWorkload(env.UpStreams[0], env.Dm, env.Workload, Spec.GetWorkloadOption(workload.CommandTypeRun)).
		DmWaitTaskSynced(env.DmHost, taskName, sourceNameList).
		DmCheckSyncDiff(env.Dm, env.SyncDiff, env.DmHost, Path.DiffPaths[0])
	return env.cleanUpEnv(dbName)
}

// cleanUpEnv when reusing testbed, clean up the testbed, currently used for local test
// may fail if the testbed is partially initialized
func (env *CommonEnv) cleanUpEnv(dbName string) *CommonEnv {
	for _, taskPath := range Path.TaskPaths {
		env.b.DmStopTask(env.Dm, env.DmHost, taskPath)
	}
	for _, sourcePath := range Path.SourcePaths {
		env.b.DmStopSource(env.Dm, env.DmHost, sourcePath)
	}
	sql := "reset master"
	dbTest := "test"
	for i := 0; i < len(env.UpStreams); i++ {
		env.b.DropDatabasesIfExists(env.UpStreams[i], &dbName)
		env.b.DropDatabasesIfExists(env.UpStreams[i], &dbTest)
		env.b.ExecuteSqlStatement(env.UpStreams[i], &sql)
	}
	env.b.DropDatabasesIfExists(env.DownStream, &dbName)
	env.b.DropDatabasesIfExists(env.DownStream, &dbTest)
	metaDb := "dm_meta"
	env.b.DropDatabasesIfExists(env.DownStream, &metaDb)
	return env
}

func (env *CommonEnv) toBuilder() *steps.Builder {
	return env.b
}

func InitCommonEnv() *CommonEnv {
	return newCommonEnv().init()
}

func (env *CommonEnv) Cleanup(dbName string) steps.Step {
	return env.cleanUpEnv(dbName).b.Build()
}