package ticdc

import (
	"bytes"
	"encoding/json"
	"fmt"

	api "github.com/pingcap/tiflow/cdc/api/owner"
	"github.com/pingcap/tiflow/cdc/model"
	"github.com/pkg/errors"
)

const configFile = "/tmp/changefeed.toml"

type Response interface {
}

type ChangeFeed struct {
	ID         string                  `json:"id"`
	Info       *model.ChangeFeedInfo   `json:"info"`
	Status     *model.ChangeFeedStatus `json:"status"`
	Count      uint64                  `json:"count"`
	TaskStatus []CaptureTaskStatus     `json:"task-status"`
	Summary    *api.ChangefeedResp     `json:"summary"`
}

type ChangeFeedList []ChangeFeed

type CaptureTaskStatus struct {
	CaptureID  string            `json:"capture-id"`
	TaskStatus *model.TaskStatus `json:"status"`
}

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

// ChangeFeedUpdateFlags can be used to update changefeed command line args.
type ChangeFeedUpdateFlags struct {
	Sink    string
	ID      string
	Config  bool
	Confirm bool
}

func (f *ChangeFeedUpdateFlags) Parse() (params []string, err error) {
	if f.ID != "" {
		params = append(params, "--changefeed-id="+f.ID)
	}
	if f.Sink != "" {
		params = append(params, "--sink-uri="+f.Sink)
	}

	params = append(params, "--no-confirm")

	// Specify changefeed.toml
	if f.Config {
		filename := configFile
		params = append(params, "--config="+filename)
	}
	return params, nil
}

func (f *ChangeFeedUpdateFlags) Unmarshal(data []byte) (Response, error) {
	index := bytes.Index(data, []byte("Update changefeed config successfully!"))
	if index == -1 {
		return nil, errors.Errorf("cannot Update changefeed successfully!")
	}

	data = data[index:]
	lines := bytes.Split(data, []byte("\n"))
	if len(lines) != 3 {
		return nil, errors.Errorf("unexpected information from update changefeed, info=%+v", data)
	}
	if !bytes.HasPrefix(lines[1], []byte("ID: ")) {
		return nil, errors.Errorf("unexpected information from update changefeed, info=%+v", data)
	}
	id := bytes.TrimPrefix(lines[1], []byte("ID: "))

	if !bytes.HasPrefix(lines[2], []byte("Info: ")) {
		return nil, errors.Errorf("unexpected information from update changefeed, info=%+v", data)
	}
	info := bytes.TrimPrefix(lines[2], []byte("Info: "))

	changefeedInfo := &model.ChangeFeedInfo{}
	if err := json.Unmarshal(info, changefeedInfo); err != nil {
		return nil, errors.New("failed to unmarshal info")
	}

	return &ChangeFeed{
		ID:   string(id),
		Info: changefeedInfo,
	}, nil
}

type SinkType int

const (
	SinkTypeMySQL = iota
	SinkTypeKafka
)

func (t SinkType) String() string {
	switch t {
	case SinkTypeMySQL:
		return "MySQL"
	case SinkTypeKafka:
		return "Kafka"
	}
	panic("invalid sink type")
}

// ChangeFeedCreateFlags can be used to create changefeed command line args.
type ChangeFeedCreateFlags struct {
	SinkType                SinkType `json:"sink-type"`
	ID                      string   `json:"changefeed-id"`
	StartTs                 uint64   `json:"start_ts"`
	NoConfirm               bool     `json:"no-confirm"`
	DisableGCSafePointCheck bool     `json:"disable_gc_safe_point_check"`
	Config                  bool
	ConfigFile              string `json:"config_file"`

	// Options can be used to format SinkURI, it's different by sink type.
	Options map[string]string `json:"options"`
}

func (f *ChangeFeedCreateFlags) Parse() (params []string, err error) {
	if f.ID != "" {
		params = append(params, "--changefeed-id="+f.ID)
	}
	if f.StartTs > 0 {
		params = append(params, fmt.Sprintf("--start-ts=%d", f.StartTs))
	}
	if f.DisableGCSafePointCheck {
		params = append(params, "--disable-gc-check")
	}

	if f.ConfigFile != "" {
		params = append(params, "--config="+f.ConfigFile)
	}

	// Specify changefeed.toml
	if f.Config {
		filename := configFile
		params = append(params, "--config="+filename)
	}

	params = append(params, "--no-confirm")

	return params, nil
}

func (f *ChangeFeedCreateFlags) Unmarshal(data []byte) (Response, error) {
	index := bytes.Index(data, []byte("Create changefeed successfully!"))
	if index == -1 {
		return nil, errors.Errorf("cannot find Create changefeed successfully!")
	}

	data = data[index:]
	lines := bytes.Split(data, []byte("\n"))
	if len(lines) != 3 {
		return nil, errors.Errorf("unexpected information from create changefeed, info=%+v", data)
	}
	if !bytes.HasPrefix(lines[1], []byte("ID: ")) {
		return nil, errors.Errorf("unexpected information from create changefeed, info=%+v", data)
	}
	id := bytes.TrimPrefix(lines[1], []byte("ID: "))

	if !bytes.HasPrefix(lines[2], []byte("Info: ")) {
		return nil, errors.Errorf("unexpected information from create changefeed, info=%+v", data)
	}
	info := bytes.TrimPrefix(lines[2], []byte("Info: "))

	changefeedInfo := &model.ChangeFeedInfo{}
	if err := json.Unmarshal(info, changefeedInfo); err != nil {
		return nil, errors.New("failed to unmarshal info")
	}

	return &ChangeFeed{
		ID:   string(id),
		Info: changefeedInfo,
	}, nil
}

type ChangeFeedQueryFlags struct {
	Simple bool
	ID     string
}

func (f *ChangeFeedQueryFlags) Parse() (params []string, err error) {
	if f.Simple {
		params = append(params, "-s")
	}

	params = append(params, "--changefeed-id="+f.ID)
	return params, nil
}

func (f *ChangeFeedQueryFlags) Unmarshal(data []byte) (Response, error) {
	feed := &ChangeFeed{
		ID: f.ID,
	}
	if f.Simple {
		summary := &api.ChangefeedResp{}
		if err := json.Unmarshal(data, summary); err != nil {
			return nil, err
		}
		feed.Summary = summary
		return feed, nil
	}

	if err := json.Unmarshal(data, feed); err != nil {
		return nil, err
	}
	return feed, nil
}

type ChangeFeedListFlags struct {
}

func (f *ChangeFeedListFlags) Parse() (params []string, err error) {
	return params, err
}

func (f *ChangeFeedListFlags) Unmarshal(data []byte) (Response, error) {
	var result ChangeFeedList
	if err := json.Unmarshal(data, &result); err != nil {
		return nil, err
	}
	return result, nil
}

type ChangeFeedAdminJobFlags struct {
	JobType     model.AdminJobType
	ID          string
	ForceRemove bool
	NoConfirm   bool
}

func (f *ChangeFeedAdminJobFlags) Parse() (params []string, err error) {
	if f.ID == "" {
		return nil, fmt.Errorf("ID is not set")
	}
	params = append(params, "--changefeed-id="+f.ID)
	// if f.ForceRemove && f.JobType == model.AdminRemove {
	// 	params = append(params, "-f")
	// }
	if f.NoConfirm && f.JobType == model.AdminResume {
		params = append(params, "--no-confirm")
	}
	return params, nil

}

func (f *ChangeFeedAdminJobFlags) Unmarshal(data []byte) (Response, error) {
	return nil, nil
}