//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 = &gcv2Worker{}

const (
	gcv2PollInterval = 1 * time.Minute
	gcv2PollTimeout  = 30 * time.Minute
)

type gcv2Worker struct {
	wctx   *workerContext
	logger logr.Logger

	jobID     int64
	elementID int64
}

func newGCV2Worker(f *framework.Framework, wctx *workerContext) *gcv2Worker {
	logger := f.Logger.WithName("gcv2")
	return &gcv2Worker{
		wctx:   wctx,
		logger: logger,
	}
}

func (w *gcv2Worker) updateTaskInfo(jobID, elementID int64) {
	w.jobID = jobID
	w.elementID = elementID
}

func (w *gcv2Worker) name() string {
	return "gcv2"
}

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

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

	_, err = w.wctx.db.ExecContext(ctx, `drop table test.gcv2_t`)
	if err != nil {
		return err
	}

	sql := `select job_id, element_id from mysql.gc_delete_range order by ts desc limit 1`
	rows, err := w.wctx.adminDB.QueryContext(ctx, sql)
	if err != nil {
		return err
	}
	if !rows.Next() {
		return errors.Errorf("unexpected result")
	}
	if rows.Err() != nil {
		return rows.Err()
	}

	var (
		jobID     int64
		elementID int64
	)
	err = rows.Scan(&jobID, &elementID)
	if err != nil {
		return err
	}

	w.updateTaskInfo(jobID, elementID)
	w.logger.Info("trigger gcv2 job", "job id", jobID, "element id", elementID)
	return nil
}

func (w *gcv2Worker) verifyJob(ctx context.Context) error {
	ticker := time.NewTicker(gcv2PollInterval)
	timer := time.NewTicker(gcv2PollTimeout)
	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 count(*) from mysql.gc_delete_range_done where job_id = ? and element_id = ?`
			rows, err := w.wctx.adminDB.QueryContext(ctx, sql, w.jobID, w.elementID)
			if err != nil || !rows.Next() {
				continue
			}
			if rows.Err() != nil {
				return rows.Err()
			}
			var count int64
			err = rows.Scan(&count)
			if err != nil {
				return err
			}
			if count == 1 {
				return nil
			}
		case <-timer.C:
			return errors.Errorf("waiting for gcv2 job timeout")
		}
	}
}