//nolint:unused,dupl // skip for unstable test, for readability.
package worker

import (
	"context"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/tests/framework"
)

var _ worker = &bgtaskWorker{}

const (
	batchPollInterval = 1 * time.Minute
	batchPollTimeout  = 10 * time.Minute
)

type bgtaskWorker struct {
	wctx   *workerContext
	logger logr.Logger

	taskID  int64
	taskKey string
}

func newBgtaskWorker(f *framework.Framework, wctx *workerContext) *bgtaskWorker {
	logger := f.Logger.WithName("bgtask")
	return &bgtaskWorker{
		wctx:   wctx,
		logger: logger,
	}
}

func (w *bgtaskWorker) updateTaskInfo(taskID int64, taskKey string) {
	w.taskID = taskID
	w.taskKey = taskKey
}

func (w *bgtaskWorker) name() string {
	return "bgtask"
}

func (w *bgtaskWorker) triggerJob(ctx context.Context) error {
	_, err := w.wctx.db.ExecContext(ctx, `create table test.batch_t1(a int primary key, b int, c int)`)
	if err != nil {
		return err
	}

	_, err = w.wctx.db.ExecContext(ctx, `insert into test.batch_t1 value(1, 2, 3)`)
	if err != nil {
		return err
	}

	_, err = w.wctx.db.ExecContext(ctx, `create table test.batch_t2(a int primary key, b int, c int)`)
	if err != nil {
		return err
	}

	// trigger batch job
	sql := `batch on test.batch_t1.a limit 10000 async insert into test.batch_t2 select * from test.batch_t1`
	rows, err := w.wctx.db.QueryContext(ctx, sql)
	if err != nil {
		return err
	}
	if !rows.Next() {
		return errors.Errorf("unexpected result")
	}
	if rows.Err() != nil {
		return rows.Err()
	}

	var (
		id       int64
		task_key string
	)
	err = rows.Scan(&id, &task_key)
	if err != nil {
		return err
	}

	w.updateTaskInfo(id, task_key)
	w.logger.Info("trigger batch job", "task id", id, "task key", task_key)
	return nil
}

func (w *bgtaskWorker) verifyJob(ctx context.Context) error {
	ticker := time.NewTicker(batchPollInterval)
	timer := time.NewTicker(batchPollTimeout)
	now := time.Now()

	defer func() {
		ticker.Stop()
		timer.Stop()
		w.logger.Info("verifyJob done", "takes", time.Since(now))
	}()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			sql := `select state from mysql.tidb_global_task_history where id = ? and task_key = ?`
			rows, err := w.wctx.db.QueryContext(ctx, sql, w.taskID, w.taskKey)
			if err != nil || !rows.Next() {
				continue
			}
			if rows.Err() != nil {
				return rows.Err()
			}
			var state string
			err = rows.Scan(&state)
			if err != nil {
				return err
			}
			if state != "succeed" {
				return errors.Errorf("unexpected state %s for batch job", state)
			}

			sql = `select count(*) from test.batch_t2`
			rows, err = w.wctx.db.QueryContext(ctx, sql)
			if err != nil {
				return err
			}
			if !rows.Next() {
				return errors.Errorf("unexpected result")
			}
			if rows.Err() != nil {
				return rows.Err()
			}
			var count int64
			err = rows.Scan(&count)
			if err != nil {
				return err
			}
			if count != 1 {
				return errors.Errorf("unexpected result")
			}
			return nil
		case <-timer.C:
			return errors.Errorf("waiting for batch job timeout")
		}
	}
}