package steps

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap/test-infra/caselib/pkg/host"
	dmmodel "github.com/pingcap/test-infra/caselib/pkg/model/dm"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
)

type operateTaskStep struct {
	dmHost   *host.DMHost
	taskPath string
	op       dmmodel.OperateTaskType
}

func (st *operateTaskStep) Execute(ctx context.Context) error {
	err := st.dmHost.OperateTask(&dmmodel.TaskOption{
		ConfigPath:  st.taskPath,
		OperateType: st.op,
	})
	if err != nil {
		return err
	}
	return nil
}

func (st *operateTaskStep) String() string {
	return st.op.String() + " by task file: " + st.taskPath
}

func (b *Builder) DmStartTask(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPath string) *Builder {
	b.steps = append(b.steps, &operateTaskStep{
		dmHost:   dmHost,
		taskPath: taskPath,
		op:       dmmodel.StartTask,
	})
	return b
}

func (b *Builder) DmStopTask(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPath string) *Builder {
	b.steps = append(b.steps, &operateTaskStep{
		dmHost:   dmHost,
		taskPath: taskPath,
		op:       dmmodel.StopTask,
	})
	return b
}

func (b *Builder) DmPauseTask(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPath string) *Builder {
	b.steps = append(b.steps, &operateTaskStep{
		dmHost:   dmHost,
		taskPath: taskPath,
		op:       dmmodel.PauseTask,
	})
	return b
}

func (b *Builder) DmResumeTask(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPath string) *Builder {
	b.steps = append(b.steps, &operateTaskStep{
		dmHost:   dmHost,
		taskPath: taskPath,
		op:       dmmodel.ResumeTask,
	})
	return b
}

func (b *Builder) DmStartTasks(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPaths []string) *Builder {
	for _, taskPath := range taskPaths {
		b.DmStartTask(dm, dmHost, taskPath)
	}
	return b
}

type waitTaskSynced struct {
	dmHost         *host.DMHost
	taskName       string
	sourceNameList []string
}

// NewWaitTaskSyncedStep create a NewWaitTaskSyncedStep which will wait the task synced
func NewWaitTaskSyncedStep(dmHost *host.DMHost, taskName string, sourceNameList []string) *waitTaskSynced {
	return &waitTaskSynced{
		dmHost:         dmHost,
		taskName:       taskName,
		sourceNameList: sourceNameList,
	}
}

func (w waitTaskSynced) String() string {
	return "waitTaskSynced " + w.taskName
}

func (w waitTaskSynced) Execute(ctx context.Context) error {
	// wait at most for 5 minutes, if the task is still not synced, return error
	for i := 0; i < 100; i++ {
		_, status, err := w.dmHost.QueryStatus(w.taskName, w.sourceNameList...)
		if err != nil {
			return err
		}
		allSynced := true
		for _, source := range status.Sources {
			for _, subtask := range source.GetSubTaskStatus() {
				if s := subtask.GetSync(); s == nil || !s.Synced {
					// not all subtasks have sync status or not synced, so we need to wait
					allSynced = false
					break
				}
			}
		}
		if allSynced {
			log.Info("waitTaskSynced all subtask synced", zap.String("task", w.taskName))
			return nil
		}
		log.Info(fmt.Sprintf("waitTaskSynced not all subtask synced for %d time", i+1), zap.String("task", w.taskName))
		time.Sleep(time.Second * 6)
	}
	return fmt.Errorf("waitTaskSynced failed")
}

func (b *Builder) DmWaitTaskSynced(dmHost *host.DMHost, taskName string, sourceNameList []string) *Builder {
	b.steps = append(b.steps, NewWaitTaskSyncedStep(dmHost, taskName, sourceNameList))
	return b
}

type waitLongDurationTaskSynced struct {
	dmHost         *host.DMHost
	taskName       string
	sourceNameList []string
}

// NewWaitTaskSyncedStep create a NewWaitTaskSyncedStep which will wait the task synced
func NewWaitLongDurationTaskSyncedStep(dmHost *host.DMHost, taskName string, sourceNameList []string) *waitLongDurationTaskSynced {
	return &waitLongDurationTaskSynced{
		dmHost:         dmHost,
		taskName:       taskName,
		sourceNameList: sourceNameList,
	}
}

func (w waitLongDurationTaskSynced) String() string {
	return "waitLongDurationTaskSynced " + w.taskName
}

func (w waitLongDurationTaskSynced) Execute(ctx context.Context) error {
	// wait at most for 24 hour, if the task is still not synced, return error
	for i := 0; i < 500; i++ {
		_, status, err := w.dmHost.QueryStatus(w.taskName, w.sourceNameList...)
		if err != nil {
			return err
		}
		allSynced := true
		for _, source := range status.Sources {
			for _, subtask := range source.GetSubTaskStatus() {
				if s := subtask.GetSync(); s == nil || !s.Synced {
					// not all subtasks have sync status or not synced, so we need to wait
					allSynced = false
					break
				}
			}
		}
		if allSynced {
			log.Info("waitTaskSynced all subtask synced", zap.String("task", w.taskName))
			return nil
		}
		log.Info(fmt.Sprintf("waitTaskSynced not all subtask synced for %d time", i+1), zap.String("task", w.taskName))
		time.Sleep(time.Second * 1000)
	}
	return fmt.Errorf("waitTaskSynced failed")
}

func (b *Builder) DmWaitLongDurationTaskSynced(dmHost *host.DMHost, taskName string, sourceNameList []string) *Builder {
	b.steps = append(b.steps, NewWaitLongDurationTaskSyncedStep(dmHost, taskName, sourceNameList))
	return b
}

// Tiup start task
type tiupDmStartTask struct {
	dmHost   *host.DMHost
	taskPath string
	// sourceNameList []string
}

func (w tiupDmStartTask) String() string {
	return "tiupDmStartTask " + w.taskPath
}

func (w tiupDmStartTask) Execute(ctx context.Context) error {
	// wait at most for 5 minutes, if the task is still not synced, return error
	tmpPath := utils.TempPath(w.taskPath)
	err := w.dmHost.TiupStartTask(tmpPath)
	if err != nil {
		return err
	}
	return nil
}

func NewTiupDmStartTask(dmHost *host.DMHost, taskPath string) *tiupDmStartTask {
	return &tiupDmStartTask{
		dmHost:   dmHost,
		taskPath: taskPath,
	}
}

func (b *Builder) TiupDmStartTasks(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPaths []string) *Builder {
	for _, taskPath := range taskPaths {
		b.TiupDmStartTask(dm, dmHost, taskPath)
	}
	return b
}

func (b *Builder) TiupDmStartTask(dm *apiresource.ResourceHandle, dmHost *host.DMHost, taskPath string) *Builder {
	b.steps = append(b.steps, NewTiupDmStartTask(dmHost, taskPath))
	return b
}