package steps

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/host"
	"github.com/pingcap/test-infra/caselib/pkg/model/ticdc"
	"go.uber.org/zap"
)

type changeFeedTask struct {
	flag ticdc.ChangeFeedParams
}

func NewChangeFeedTask(flag ticdc.ChangeFeedParams) Step {
	return &changeFeedTask{
		flag: flag,
	}
}

func GetCDCHost(ctx context.Context) (*host.TiCDCHost, error) {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return nil, errors.Trace(err)
	}

	sink, err := GetInner(ctx).Board().Sink()
	if err != nil {
		return nil, errors.Trace(err)
	}

	cdcHost, err := host.NewTiCDCHost(tc, sink)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return cdcHost, nil
}

func (s *changeFeedTask) Execute(ctx context.Context) error {
	cdcHost, err := GetCDCHost(ctx)
	if err != nil {
		return err
	}

	result, err := cdcHost.ChangeFeed(s.flag)
	log.Info("run changeFeed task finished", zap.Any("result", result), zap.Error(err))
	return err
}

func (s *changeFeedTask) String() string {
	return fmt.Sprintf("changeFeedTask: flag=%+v", s.flag)
}