package ticdc

import (
	"bytes"
	"regexp"
	"strconv"
	"strings"

	"github.com/pkg/errors"
)

type RedoParams interface {
	Parse() ([]string, error)
	Unmarshal(stdout []byte) (Response, error)
}

type RedoApplyFlags struct {
	Storage string
	TmpDir  string
}

func (f *RedoApplyFlags) Parse() (params []string, err error) {
	params = append(params, "--storage="+f.Storage)
	params = append(params, "--tmp-dir=/tmp/nfs/"+f.TmpDir)

	return params, nil
}

func (f *RedoApplyFlags) Unmarshal(data []byte) (Response, error) {
	index := bytes.Index(data, []byte("Apply redo log successfully"))
	if index == -1 {
		return nil, errors.Errorf("cannot apply redo log successfully")
	}

	return string(data), nil
}

type RedoMetaFlags struct {
	Storage string
	TmpDir  string
}

func (f *RedoMetaFlags) Parse() (params []string, err error) {
	params = append(params, "--storage="+f.Storage)
	params = append(params, "--tmp-dir=/tmp/nfs/meta/"+f.TmpDir)

	return params, nil
}

func parseRedoMetaFromLines(lines []string) (meta *RedoMeta, err error) {
    var checkpointTsPattern, resolvedTsPattern *regexp.Regexp
    checkpointTsPattern, _ = regexp.Compile("checkpoint-ts:([0-9]+)")
    resolvedTsPattern, _ = regexp.Compile("resolved-ts:([0-9]+)")
    meta = &RedoMeta{}

    for _, line := range lines {
        if checkpointTsPattern == nil && resolvedTsPattern == nil {
            break
        }
        if checkpointTsPattern != nil {
            matches := checkpointTsPattern.FindStringSubmatch(line)
            if len(matches) == 2 {
                x, err := strconv.Atoi(matches[1])
                if err != nil {
                    return nil, err
                }
                meta.CheckPointTs = uint64(x)
                checkpointTsPattern = nil
            }
        }
        if resolvedTsPattern != nil {
            matches := resolvedTsPattern.FindStringSubmatch(line)
            if len(matches) == 2 {
                x, err := strconv.Atoi(matches[1])
                if err != nil {
                    return nil, err
                }
                meta.ResolvedTs = uint64(x)
                resolvedTsPattern = nil
            }
        }
    }
    if resolvedTsPattern != nil || checkpointTsPattern != nil {
        err = errors.Errorf("unexpected information from redo meta, info=%s", strings.Join(lines, "\n"))
    }
    return
}

func (f *RedoMetaFlags) Unmarshal(data []byte) (Response, error) {
	// checkpoint-ts:429327239870152705, resolved-ts:429327242295246851
	lines := strings.Split(string(data), "\n")
    return parseRedoMetaFromLines(lines)
}

type RedoMeta struct {
	CheckPointTs uint64 `json:"checkPointTs"`
	ResolvedTs   uint64 `json:"resolvedTs"`
}