package main

import (
	"context"
	"flag"
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"

	"github.com/ngaut/log"
	client "github.com/pingcap/schrodinger/pkg/test_client"
)

var (
	mysqlFlavor       string
	mysqlVersion      string
	dmWorkerCount     int
	dmCaseCount       int
	downloadAddr      string
	ddlCount          int
	dmWorkerMem       int
	withNemesis       bool
	withCheck         bool
	taskMode          string
	onlineDDL         bool
	onlineDDLTool     string
	filterDDL         bool
	cmptDDL           bool
	masterSlaveSwitch bool
	failResume        bool
	enableGTID        bool
	maxBinlogSize     int64
)

var (
	AllMode  = "all"
	FullMode = "full"
)

func init() {
	flag.StringVar(&mysqlFlavor, "mysql-flavor", "mysql", "mysql flavor (mysql or mariadb)")
	flag.StringVar(&mysqlVersion, "mysql-version", "5.7", "mysql version")
	flag.IntVar(&dmWorkerCount, "worker-count", 2, "init worker count")
	flag.IntVar(&dmCaseCount, "case-count", 2, "init case count")
	flag.IntVar(&ddlCount, "ddl-count", 2000, "how many ddl")
	flag.IntVar(&dmWorkerMem, "dm-worker-memory", 6, "dm-worker memory quota in GB")
	flag.StringVar(&downloadAddr, "download-addr", "http://download.pingcap.org/dm-latest-linux-amd64.tar.gz", "download addr")
	flag.StringVar(&taskMode, "task-mode", AllMode, "mode of test task,  full, all (= full + incremental)")
	flag.BoolVar(&onlineDDL, "online-ddl", false, "whether enable online ddl, if enabled, must use docker image: 127.0.0.1:5001/pingcap/ptagent:latest")
	flag.StringVar(&onlineDDLTool, "online-ddl-tool", "pt", "online ddl tool, pt(stand for pt-osc) or gh-ost(stand for gh-ost)")
	flag.BoolVar(&filterDDL, "filter-ddl", false, "whether run ddl in filter list")
	flag.BoolVar(&cmptDDL, "cmpt-ddl", false, "whether enable incompatible ddl")
	flag.BoolVar(&masterSlaveSwitch, "master-slave-switch", false, "whether test upstream master/slave switch")
	flag.BoolVar(&failResume, "fail-resume", false, "use fail resume test case")
	flag.BoolVar(&enableGTID, "enable-gtid", false, "enable GTID")
	flag.Int64Var(&maxBinlogSize, "max-binlog-size", 1073741824, "max binlog size")
}

func main() {

	flag.Parse()
	client.Info("[dm-test] start...")
	adjustTaskMode()

	ctx, cancel := context.WithCancel(context.Background())
	cli := client.New(ctx, client.ModeOnline)

	sc := make(chan os.Signal, 1)
	signal.Notify(sc,
		syscall.SIGHUP,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT)
	go func() {
		sig := <-sc
		client.Infof("[binlog] Got signal [%s] to exist.", sig)
		cancel()
		os.Exit(0)
	}()

	DefaultInitSqls = append(DefaultInitSqls, fmt.Sprintf("SET @@GLOBAL.MAX_BINLOG_SIZE = %d;", maxBinlogSize))

	var deploys []*Deploy
	var workers []string
	mysqls := make(map[int]*replicationMySQL)
	var err error
	for i := 0; i < dmWorkerCount; i++ {
		mysqlHosts := make([]string, 0)
		workerID := 100 + i
		if masterSlaveSwitch {
			mysqlHosts, err = setUpReplicationMySQL(cli, mysqlFlavor, mysqlVersion, workerID+i)
			if err != nil {
				client.Fatalf("[dm] setup replication mysql with error %v", client.ErrorStack(err))
			}
		} else {
			host, err := setUpMySQL(cli, mysqlFlavor, mysqlVersion, workerID)
			if err != nil {
				client.Fatalf("[dm] setup mysql with error %v", client.ErrorStack(err))
			}
			mysqlHosts = append(mysqlHosts, host)
		}
		instanceID := i + 1
		mysqls[instanceID] = &replicationMySQL{hosts: mysqlHosts, current: 0}
	}

	if masterSlaveSwitch {
		enableGTID = true
	}
	if enableGTID {
		err = setUpGTID(mysqls, masterSlaveSwitch, mysqlFlavor)
		if err != nil {
			client.Fatalf("[dm] setup GTID with error %v", client.ErrorStack(err))
		}
	}

	for i := 0; i < dmWorkerCount; i++ {
		workerID := 100 + i
		instID := i + 1
		memoryQuota := fmt.Sprintf("%dGi", dmWorkerMem)
		worker, err := setUpDmWorker(cli, workerID, mysqls[instID].hosts[0], memoryQuota)
		if err != nil {
			client.Fatalf("[dm] setup worker with error %v", client.ErrorStack(err))
		}
		deploy := &Deploy{
			SourceID: fmt.Sprintf("%s:3306", mysqls[instID].hosts[0]),
			DmWorker: fmt.Sprintf("%s:%d", worker, DMWorkerPort),
		}
		deploys = append(deploys, deploy)
		workers = append(workers, worker)
	}

	masterCfg := NewMasterConfig(deploys)
	master, err := setUpDmMaster(cli, masterCfg)
	if err != nil {
		client.Fatalf("[dm] setup master with error %v", client.ErrorStack(err))
	}

	dmctl, err := setUpDmCtl(ctx, master)
	if err != nil {
		client.Fatalf("[dm] setup master with error %v", client.ErrorStack(err))
	}

	log.SetLevelByString("info")
	client.Debugf("dmctl %+v", dmctl)

	testInfo := cli.GetTests()
	caseHostIP := testInfo[0].IP
	nemesis := newNemesis(caseHostIP, master, workers, dmctl)

	var (
		caseWg        sync.WaitGroup
		nemesisWg     sync.WaitGroup
		startupWg     sync.WaitGroup
		fullStageDone int32
		ddls          = make(map[int][]string)
	)

	// We can add multiple cases here
	for i := 0; i < dmCaseCount; i++ {
		var dmc *DMCase
		caseOpts := &caseOptions{
			ddlCount:          ddlCount,
			mysqlFlavor:       mysqlFlavor,
			taskMode:          taskMode,
			enableSharding:    false,
			masterSlaveSwitch: masterSlaveSwitch,
			cmptDDL:           cmptDDL,
			onlineDDL:         onlineDDL,
			onlineDDLTool:     onlineDDLTool,
			filterDDL:         filterDDL,
			enableGTID:        enableGTID,
			failResume:        failResume,
		}
		if i > 0 {
			// only the first case is non sharding
			caseOpts.enableSharding = true
			dmc, err = NewDMCase(ctx, i+1, master, workers, mysqls, caseOpts, nemesis)
			if err != nil {
				client.Fatalf("[dm] %s", client.ErrorStack(err))
			}
		} else {
			// non sharding case
			dmc, err = NewDMCase(ctx, i+1, master, workers, mysqls, caseOpts, nemesis)
			if err != nil {
				client.Fatalf("[dm] %s", client.ErrorStack(err))
			}
		}
		err := dmc.Init(cli, dmctl)
		if err != nil {
			client.Fatalf(client.ErrorStack(err))
		}
		// dmc.ddls will not update, so we use it directly without copy data
		ddls[dmc.caseID] = dmc.ddls
		caseWg.Add(1)
		startupWg.Add(1)
		go dmc.Run(&caseWg, &startupWg, &fullStageDone)
	}

	opts := &ControllOptions{
		mode:              taskMode,
		failResume:        failResume,
		masterSlaveSwitch: masterSlaveSwitch,
	}
	ctrl := NewDMController(ctx, nemesis, opts, ddls)
	err = ctrl.Init(cli)
	if err != nil {
		client.Fatalf(client.ErrorStack(err))
	}
	go ctrl.CheckFullStageDone(dmWorkerCount, dmCaseCount, &fullStageDone)
	startupWg.Wait()
	go ctrl.Run()

	nemesisWg.Add(1)
	go func() {
		defer nemesisWg.Done()
		select {
		case <-ctrl.ctx.Done():
			return
		case err := <-nemesis.errCh:
			client.Fatalf("[dm] %s", client.ErrorStack(err))
		}
	}()

	caseWg.Wait()
	ctrl.cancel()
	nemesisWg.Wait()
}

func adjustTaskMode() {
	if taskMode != FullMode {
		taskMode = AllMode
	}
}