package main

import (
	"context"
	"database/sql"
	"math/rand"
	"sync"
	"sync/atomic"

	"github.com/juju/errors"
	data "github.com/pingcap/schrodinger-test/pkg/ddl_data"
	client "github.com/pingcap/schrodinger/pkg/test_client"
)

type consumer struct {
	*data.Client
	db            *sql.DB
	dbStandby     *sql.DB
	dbName        string
	table         string
	dbInfo        *data.DBInfo
	standbyDBInfo *data.DBInfo
}

type executorOpts struct {
	onlineDDL     bool
	onlineDDLTool string
	filterDDL     bool
}

type executor struct {
	ctx    context.Context
	cancel context.CancelFunc
	closed int64

	consumers []*consumer
	wg        *sync.WaitGroup
	ddlCh     chan string
	errCh     chan error
	opts      *executorOpts
}

func (e *executor) addJob(ddl string) {
	if atomic.LoadInt64(&e.closed) == 1 {
		return
	}

	e.ddlCh <- ddl
}

func (e *executor) startDML() error {
	if atomic.LoadInt64(&e.closed) == 1 {
		return nil
	}

	for _, cli := range e.consumers {
		if err := cli.StartDML(cli.db, cli.dbName, cli.table); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (e *executor) close() {
	if !atomic.CompareAndSwapInt64(&e.closed, 0, 1) {
		return
	}

	e.cancel()
}

// when finish a ddl then wg.Done, run ddl line by line
func (e *executor) runDeamon() {
	for {
		select {
		case <-e.ctx.Done():
			return
		case ddl := <-e.ddlCh:
			client.Debugf("begin to execute consume ddl %s", ddl)
			done := make(chan struct{}, len(e.consumers))
			errCh := make(chan error, len(e.consumers))

			onlineDDL := false
			if e.opts.onlineDDL && rand.Intn(5) < 2 {
				onlineDDL = true
			}
			for _, cli := range e.consumers {
				args := &data.ConsumeDDLArgs{
					data.ConsumeDDLInfo{
						Db:        cli.db,
						DbInfo:    cli.dbInfo,
						DbName:    cli.dbName,
						TableName: cli.table,
						Ddl:       ddl,
					},
					data.ConsumeDDLOpts{
						Done:          done,
						ErrCh:         errCh,
						OnlineDDL:     onlineDDL,
						OnlineDDLTool: e.opts.onlineDDLTool,
					},
				}
				cli.ConsumeDDL(args)
			}

			finished := 0
			stop := false
			for !stop {
				select {
				case <-done:
					finished++
					if finished >= len(e.consumers) {
						e.wg.Done()
						client.Debugf("finish ddl %s", ddl)
						stop = true
					}
				case err := <-errCh:
					client.Errorf("run daemon error %+v", err)
					e.errCh <- err
					e.wg.Done()
					return
				}
			}
		}
	}
}

func (e *executor) swapDBStandby() (*sql.DB, *sql.DB, error) {
	for _, consumer := range e.consumers {
		consumer.db, consumer.dbStandby = consumer.dbStandby, consumer.db
		consumer.dbInfo, consumer.standbyDBInfo = consumer.standbyDBInfo, consumer.dbInfo
	}
	if len(e.consumers) > 0 {
		return e.consumers[0].db, e.consumers[0].dbStandby, nil
	}
	return nil, nil, errors.New("executor has no consumers")
}

func newExecutor(ctx context.Context, db *sql.DB, dbStandby *sql.DB, dbName string, dbInfo, standbyDBInfo *data.DBInfo, tables []string, wg *sync.WaitGroup, opts *executorOpts) *executor {
	cctx, cancel := context.WithCancel(ctx)

	e := &executor{
		ctx:    cctx,
		cancel: cancel,
		wg:     wg,
		ddlCh:  make(chan string, 200),
		opts:   opts,
	}
	for _, table := range tables {
		cli := &consumer{
			db:            db,
			dbStandby:     dbStandby,
			Client:        data.NewClient(ctx),
			dbName:        dbName,
			table:         table,
			dbInfo:        dbInfo,
			standbyDBInfo: standbyDBInfo,
		}

		e.consumers = append(e.consumers, cli)
	}

	e.errCh = make(chan error, len(e.consumers))

	go e.runDeamon()

	return e
}