package resource

import (
	"errors"
	"net/url"
	"strconv"

	"github.com/pingcap/test-infra/sdk/core"
)

// CDCSink is the interface for cdc to get sink endpoint
// Such as kafka://localhost:9092
type CDCSink interface {
	SinkEndpoint() (*url.URL, error)
}

// KafkaCluster is the kafka cluster interface
type KafkaCluster interface {
	core.Resource
	Exec(opts KafkaClusterExecOptions) (stdout string, stderr string, exitCode int, err error)
	ChaosSelect(selector KafkaClusterSelectorOpts) ([]ChaosSelector, error)
	ChaosClient() ChaosClient
	ServiceURL() (*url.URL, error)
}

// KafkaClusterSelectorOpts defines how to select kafka cluster node
type KafkaClusterSelectorOpts struct {
	Peer          int
	All           bool
	RandomSameOne bool
	Controller    bool
}

// KafkaClusterExecOptions wraps a command which wants to execute to Kafka pod
type KafkaClusterExecOptions struct {
	Command string
	Target  KafkaCompType
	Peer    int
}

const (
	Zookeeper KafkaCompType = iota
	Kafka
)

type KafkaCompType int

func (_ KafkaCompType) FromString(str string) (KafkaCompType, error) {
	switch str {
	case "zk":
		return Zookeeper, nil
	case "kafka":
		return Kafka, nil
	default:
		return -1, errors.New("unsupported KafkaCompType")
	}
}

func (t KafkaCompType) String() string {
	switch t {
	case Zookeeper:
		return "zk"
	case Kafka:
		return "kafka"
	default:
		return "#" + strconv.Itoa(int(t))
	}
}