package tikv_cdc

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"strconv"
	"strings"
	"sync"
	"time"

	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/pkg/util/pdutil"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

const (
	CdcTimeout                = 1 * time.Minute
	WaitCdcExecuteTime        = 1 * time.Second
	GCSafepointUpdateInterval = 1 * time.Minute
	CdcCheckRetry             = 3
)

type ChangefeedOp int

const (
	Create ChangefeedOp = iota
	Pause
	Resume
	Remove
	Update
)

type ChangefeedState int

const (
	Normal ChangefeedState = iota
	Stopped
	Error
	Failed
	Finished
)

type TiKVCDCRunCmd struct {
	cdcPath string
	command string
	options []string
}

type ChangefeedOptions struct {
	DstPDAddr  string
	LogFile    string
	SortEngine string
	StartTs    uint64
	TargetTs   uint64
}

func NewTiKVCDCRunCmd(cdcPath string, command string) *TiKVCDCRunCmd {
	b := &TiKVCDCRunCmd{command: command, cdcPath: cdcPath}
	return b
}

func (t *TiKVCDCRunCmd) Command(command string) *TiKVCDCRunCmd {
	t.command = command
	return t
}

func (b *TiKVCDCRunCmd) PD(pd string) *TiKVCDCRunCmd {
	b.options = append(b.options, fmt.Sprintf("--pd %s", pd))
	return b
}

func (b *TiKVCDCRunCmd) LogFile(logFile string) *TiKVCDCRunCmd {
	b.options = append(b.options, fmt.Sprintf("--log-file %s", logFile))
	return b
}

func (b *TiKVCDCRunCmd) ChangefeedId(id string) *TiKVCDCRunCmd {
	b.options = append(b.options, fmt.Sprintf("--changefeed-id %s", id))
	return b
}

func (b *TiKVCDCRunCmd) StartTs(startTs uint64) *TiKVCDCRunCmd {
	if startTs != 0 {
		b.options = append(b.options, fmt.Sprintf("--start-ts %d", startTs))
	}
	return b
}

func (b *TiKVCDCRunCmd) TargetTs(targetTs uint64) *TiKVCDCRunCmd {
	if targetTs != 0 {
		b.options = append(b.options, fmt.Sprintf("--target-ts %d", targetTs))
	}
	return b
}

func (b *TiKVCDCRunCmd) SortEngine(engine string) *TiKVCDCRunCmd {
	b.options = append(b.options, fmt.Sprintf("--sort-engine %s", engine))
	return b
}

func (b *TiKVCDCRunCmd) TiKVSinkUri(url string) *TiKVCDCRunCmd {
	url = strings.TrimPrefix(url, "http://")
	b.options = append(b.options, fmt.Sprintf("--sink-uri='tikv://%s'", url))
	return b
}

func (b *TiKVCDCRunCmd) CaptureId(captureId string) *TiKVCDCRunCmd {
	b.options = append(b.options, fmt.Sprintf("--capture-id %s", captureId))
	return b
}

func (s *TiKVCDCRunCmd) Build() string {
	return fmt.Sprintf("%s %s %s",
		s.cdcPath, s.command, strings.Join(s.options[:], " "))
}

type TiKVCDCClient struct {
	Cdc     resource.WorkloadNode
	CdcPath string
	PDAddr  string
}

func NewTiKVCDCClient(cdc resource.WorkloadNode, cdcPath string, pdAddr string) *TiKVCDCClient {
	client := &TiKVCDCClient{
		Cdc:     cdc,
		CdcPath: cdcPath,
		PDAddr:  pdAddr,
	}
	return client
}

func (c *TiKVCDCClient) TiKVCDCCreateChangefeed(ctx context.Context,
	changefeedId string, options *ChangefeedOptions) error {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed create")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		SortEngine(options.SortEngine).
		TiKVSinkUri(options.DstPDAddr).
		StartTs(options.StartTs).
		TargetTs(options.TargetTs).
		Build()

	_, err := ExecWorkloadCmd(ctxcdc, c.Cdc, tikvCdcCmdStr, true, nil)
	return errors.Trace(err)
}

func (c *TiKVCDCClient) TiKVCDCRemoveChangefeed(ctx context.Context, changefeedId string) error {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed remove")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		Build()
	_, err := ExecWorkloadCmd(ctxcdc, c.Cdc, tikvCdcCmdStr, true, nil)
	return errors.Trace(err)
}

func (c *TiKVCDCClient) TiKVCDCQueryChangefeed(ctx context.Context, changefeedId string) (*ChangefeedMeta, error) {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed query")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		Build()

	output, _, err := ExecTiKVCmdWithOutput(ctxcdc, c.Cdc, tikvCdcCmdStr, true)
	if err != nil {
		return nil, errors.Trace(err)
	}
	cfMeta := &ChangefeedMeta{}
	err = json.Unmarshal(output, cfMeta)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return cfMeta, nil
}

func (c *TiKVCDCClient) TiKVCDCPauseChangefeed(ctx context.Context, changefeedId string) error {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed pause")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		Build()
	_, err := ExecWorkloadCmd(ctxcdc, c.Cdc, tikvCdcCmdStr, true, nil)
	return errors.Trace(err)
}

func (c *TiKVCDCClient) TiKVCDCResumeChangefeed(ctx context.Context, changefeedId string) error {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed resume")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		Build()
	_, err := ExecWorkloadCmd(ctxcdc, c.Cdc, tikvCdcCmdStr, true, nil)
	return errors.Trace(err)
}

func (c *TiKVCDCClient) TiKVCDCUpdateChangefeed(ctx context.Context, changefeedId string, options *ChangefeedOptions) error {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed update")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		SortEngine(options.SortEngine).
		TiKVSinkUri(options.DstPDAddr).
		StartTs(options.StartTs).
		TargetTs(options.TargetTs).
		Build()
	_, err := ExecWorkloadCmd(ctxcdc, c.Cdc, tikvCdcCmdStr, true, nil)
	return errors.Trace(err)
}

func (c *TiKVCDCClient) TiKVCDCStatisticsChangefeed(ctx context.Context, cdc resource.WorkloadNode,
	cdcPath string, pdAddr string, changefeedId string, output chan StatisticsInfo) error {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(cdcPath, "cli changefeed statistics")
	tikvCdcCmdStr := tikvCdcCmd.PD(pdAddr).
		ChangefeedId(changefeedId).
		Build()
	_, reader, err := ExecTiKVCmdWithOutput(ctxcdc, cdc, tikvCdcCmdStr, false)
	if err != nil {
		return errors.Trace(err)
	}
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			buf := make([]byte, 128)
			n, err := reader.Read(buf)
			if err == io.EOF {
				return nil
			}
			if err != nil {
				return errors.Trace(err)
			}
			buf = buf[:n]
			info := StatisticsInfo{}
			err = json.Unmarshal(buf, &info)
			if err != nil {
				return errors.Trace(err)
			}
			select {
			case <-ctx.Done():
				return nil
			case output <- info:
			}
		}
	}
}

func (c *TiKVCDCClient) TiKVCDCCaptureList(ctx context.Context) ([]CaptureInfo, error) {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli capture list")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).Build()
	output, _, err := ExecTiKVCmdWithOutput(ctxcdc, c.Cdc, tikvCdcCmdStr, true)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var captureInfos []CaptureInfo
	err = json.Unmarshal(output, &captureInfos)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return captureInfos, nil
}

func (c *TiKVCDCClient) TiKVCDCProcessorList(ctx context.Context) ([]ProcessorInfo, error) {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli processor list")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).Build()

	output, _, err := ExecTiKVCmdWithOutput(ctxcdc, c.Cdc, tikvCdcCmdStr, true)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var processorInfos []ProcessorInfo
	err = json.Unmarshal(output, &processorInfos)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return processorInfos, nil
}

func (c *TiKVCDCClient) TiKVCDCProcessorQuery(ctx context.Context,
	changefeedId string, captureId string) (*ProcessorMeta, error) {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli processor query")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).
		ChangefeedId(changefeedId).
		CaptureId(captureId).
		Build()
	output, _, err := ExecTiKVCmdWithOutput(ctxcdc, c.Cdc, tikvCdcCmdStr, true)
	if err != nil {
		return nil, errors.Trace(err)
	}
	processorMeta := &ProcessorMeta{}
	err = json.Unmarshal(output, processorMeta)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return processorMeta, err
}

func (c *TiKVCDCClient) TiKVCDCTsoQuery(ctx context.Context) (uint64, error) {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli tso query")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).Build()
	output, _, err := ExecTiKVCmdWithOutput(ctxcdc, c.Cdc, tikvCdcCmdStr, true)
	if err != nil {
		return 0, errors.Trace(err)
	}

	return strconv.ParseUint(string(output[:len(output)-1]), 10, 64)
}

func (c *TiKVCDCClient) TiKVCDCChangefeedList(ctx context.Context) ([]ChangefeedCommonInfo, error) {
	ctxcdc, cancel := context.WithTimeout(ctx, CdcTimeout)
	defer cancel()

	tikvCdcCmd := NewTiKVCDCRunCmd(c.CdcPath, "cli changefeed list")
	tikvCdcCmdStr := tikvCdcCmd.PD(c.PDAddr).Build()

	output, _, err := ExecTiKVCmdWithOutput(ctxcdc, c.Cdc, tikvCdcCmdStr, true)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var cfInfos []ChangefeedCommonInfo
	err = json.Unmarshal(output, &cfInfos)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return cfInfos, nil
}

func (c *TiKVCDCClient) GetTiKVCDCChangefeedInfo(ctx context.Context, changefeedId string) (*ChangefeedCommonInfo, error) {
	cfInfos, err := c.TiKVCDCChangefeedList(ctx)
	if err != nil {
		return nil, errors.Trace(err)
	}
	for _, feedInfo := range cfInfos {
		if feedInfo.ID == changefeedId {
			log.Info("Get changefeed state", zap.String("changefeed_id", changefeedId),
				zap.String("state", feedInfo.Summary.FeedState))
			return &feedInfo, nil
		}
	}
	return nil, errors.Errorf("Not found feed: %s", changefeedId)
}

func (c *TiKVCDCClient) TiKVCDCMustCreateChangefeed(ctx context.Context, changefeedId string, options *ChangefeedOptions) error {
	err := c.TiKVCDCCreateChangefeed(ctx, changefeedId, options)
	if err != nil {
		return errors.Trace(err)
	}
	return c.ChangefeedStateCheck(ctx, changefeedId, "normal")
}

func (c *TiKVCDCClient) TiKVCDCMustPauseChangefeed(ctx context.Context,
	changefeedId string) error {
	err := c.TiKVCDCPauseChangefeed(ctx, changefeedId)
	if err != nil {
		return errors.Trace(err)
	}
	return c.ChangefeedStateCheck(ctx, changefeedId, "stopped")
}

func (c *TiKVCDCClient) TiKVCDCMustResumeChangefeed(ctx context.Context,
	changefeedId string) error {
	err := c.TiKVCDCResumeChangefeed(ctx, changefeedId)
	if err != nil {
		return errors.Trace(err)
	}

	return c.ChangefeedStateCheck(ctx, changefeedId, "normal")
}

func (c *TiKVCDCClient) TiKVCDCMustRemoveChangefeed(ctx context.Context, changefeedId string) (bool, error) {
	cfInfo, _ := c.GetTiKVCDCChangefeedInfo(ctx, changefeedId)
	ok := false
	if cfInfo != nil {
		ok = true
		err := c.TiKVCDCRemoveChangefeed(ctx, changefeedId)
		if err != nil {
			return ok, errors.Trace(err)
		}
		return ok, c.ChangefeedStateCheck(ctx, changefeedId, "")
	}
	return ok, nil
}

func (c *TiKVCDCClient) ChangefeedStateCheck(ctx context.Context,
	changefeedId string, state string) error {
	err := Retry(CdcCheckRetry, WaitCdcExecuteTime, func() (bool, error) {
		time.Sleep(WaitCdcExecuteTime)
		cfInfo, err := c.GetTiKVCDCChangefeedInfo(ctx, changefeedId)
		if cfInfo == nil && state == "" {
			return true, nil
		}
		if err != nil {
			return false, errors.Trace(err)
		}
		return cfInfo.Summary.FeedState == state, nil
	})
	if err != nil {
		log.Warn("Failed to check changefeed", zap.String("expected state", state))
		return errors.Errorf("Failed to check changefeed state, expected state: %s, err: %v", state, err)
	}
	return nil
}

type ChangefeedRunner struct {
	Client       *TiKVCDCClient
	ChangefeedId string
	Options      *ChangefeedOptions

	opCh   chan ChangefeedOp
	doneCh chan struct{}
	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup
}

func NewChangefeedRunner(ctx context.Context, client *TiKVCDCClient,
	changefeedId, dstPDAddr, sortEngine string) (*ChangefeedRunner, error) {
	options := &ChangefeedOptions{
		DstPDAddr:  dstPDAddr,
		SortEngine: sortEngine,
	}
	nctx, cancel := context.WithCancel(ctx)
	cfRunner := &ChangefeedRunner{
		Client:       client,
		ChangefeedId: changefeedId,
		Options:      options,

		opCh:   make(chan ChangefeedOp),
		doneCh: make(chan struct{}),
		ctx:    nctx,
		cancel: cancel,
		wg:     sync.WaitGroup{},
	}
	err := cfRunner.StartChangefeedRunner(cfRunner.ctx)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return cfRunner, nil
}

func (cfRunner *ChangefeedRunner) Close() {
	cfRunner.cancel()
	cfRunner.wg.Wait()
}

func (r *ChangefeedRunner) StartChangefeedRunner(ctx context.Context) error {
	// try to remove last changefeed
	_, err := r.Client.TiKVCDCMustRemoveChangefeed(ctx, r.ChangefeedId)
	if err != nil {
		return errors.Trace(err)
	}
	go r.Runner(ctx)
	return nil
}

func (r *ChangefeedRunner) Runner(ctx context.Context) {
	r.wg.Add(1)
	defer r.wg.Done()

	changefeedId := r.ChangefeedId
	options := r.Options
	client := r.Client
	state := Finished
	defer func() {
		if state != Finished {
			log.Info("Try to remove changefeed", zap.String("changefeedId", changefeedId))
			_, err := client.TiKVCDCMustRemoveChangefeed(ctx, changefeedId)
			if err != nil {
				log.Warn("Could not remove changefeed", zap.String("changefeedId", changefeedId), zap.Error(err))
			}
		}
		log.Info("Exit TiKVRunner")
	}()

	for {
		select {
		case op := <-r.opCh:
			switch op {
			case Create:
				Expect(state).To(Equal(Finished))
				err := client.TiKVCDCMustCreateChangefeed(ctx, changefeedId, options)
				Expect(err).Should(Succeed())
				state = Normal
			case Pause:
				Expect(state).To(Equal(Normal))
				err := client.TiKVCDCMustPauseChangefeed(ctx, changefeedId)
				Expect(err).Should(Succeed())
				state = Stopped
			case Resume:
				Expect(state).To(Equal(Stopped))
				err := client.TiKVCDCMustResumeChangefeed(ctx, changefeedId)
				Expect(err).Should(Succeed())
				state = Normal
			case Remove:
				Expect(state).To(Not(Equal(Finished)))
				_, err := client.TiKVCDCMustRemoveChangefeed(ctx, changefeedId)
				Expect(err).Should(Succeed())
				state = Finished
			case Update:
				Expect(state).To(Not(Equal(Finished)))
				err := client.TiKVCDCUpdateChangefeed(ctx, changefeedId, options)
				Expect(err).Should(Succeed())
			}
			select {
			case r.doneCh <- struct{}{}:
			case <-ctx.Done():
				return
			}
		case <-ctx.Done():
			return
		}
	}
}

func (r *ChangefeedRunner) ChangefeedOpHelper(ctx context.Context, op ChangefeedOp) {
	select {
	case <-ctx.Done():
	case r.opCh <- op:
		log.Info("Start changefeed op", zap.Any("op", op))
	}
	select {
	case <-ctx.Done():
	case <-r.doneCh:
		log.Info("Finish changfeed op", zap.Any("op", op))
	}
}

func (r *ChangefeedRunner) StartSafePointCheck(checkpoint uint64, pdCli *pdutil.Client) {
	go r.TiKVCDCSafePointCheck(r.ctx, checkpoint, pdCli)
}

func (r *ChangefeedRunner) TiKVCDCSafePointCheck(ctx context.Context,
	checkpoint uint64, pdCli *pdutil.Client) {
	r.wg.Add(1)
	defer r.wg.Done()

	checkSafePoint := func(checkpoint uint64) error {
		err := Retry(CdcCheckRetry, GCSafepointUpdateInterval/2, func() (bool, error) {
			cdcSafePoint, err := GetServiceSafePoint(pdCli, TiKVCDCServiceSafePointID)
			if err != nil {
				return false, err
			}
			log.Info("Service safe point should be greater than or equal to checkpoint - 1", zap.Uint64("cdcSafePoint", cdcSafePoint), zap.Uint64("checkpoint", checkpoint))
			return cdcSafePoint >= (checkpoint - 1), nil
		})
		if err != nil {
			return errors.Errorf("Service safe point should be greater than or equal to checkpoint - 1")
		}
		return nil
	}

	time.Sleep(WaitCdcExecuteTime)
	err := checkSafePoint(checkpoint)
	Expect(err).Should(Succeed())

	for {
		select {
		case <-time.After(GCSafepointUpdateInterval):
			cfInfo, err := r.Client.GetTiKVCDCChangefeedInfo(ctx, r.ChangefeedId)
			Expect(err).Should(Succeed())
			ncheckpoint := cfInfo.Summary.TSO
			Expect(ncheckpoint).Should(BeNumerically(">=", checkpoint))

			checkpoint = ncheckpoint
			err = checkSafePoint(ncheckpoint)
			select {
			case <-ctx.Done():
				return
			default:
				Expect(err).Should(Succeed())
			}
		case <-ctx.Done():
			return
		}
	}
}

func Retry(retry int, sleepTime time.Duration, f func() (bool, error)) error {
	for retry > 0 {
		ok, err := f()
		if err != nil {
			return errors.Trace(err)
		}
		if ok {
			break
		}
		time.Sleep(sleepTime)
		retry--
	}
	if retry == 0 {
		return errors.Errorf("Failed to retry, retry %d, sleep time %v", retry, sleepTime)
	}
	return nil
}

func ExecTiKVCmdWithOutput(ctx context.Context, cdc resource.WorkloadNode, cmd string, readEOF bool) ([]byte, *io.PipeReader, error) {
	cmdpr, cmdpw := io.Pipe()
	go func() {
		_, err := ExecWorkloadCmd(ctx, cdc, cmd, true, cmdpw)
		if err != nil {
			cmdpw.CloseWithError(err)
		} else {
			cmdpw.Close()
		}
	}()

	buffer := new(bytes.Buffer)
	if readEOF {
		_, err := buffer.ReadFrom(cmdpr)
		if err != nil {
			return nil, nil, err
		}
	}
	return buffer.Bytes(), cmdpr, nil
}