package api

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	cspb "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/service-proxy/pkg/checker/core"
	"go.opentelemetry.io/otel"

	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/client"
)

const (
	cdcRoutePath = "/ticdc"
	cdcPathV2    = "/api/v2/changefeeds"
	checkPath    = "/check"

	defaultTiDBName = "db"

	basePath = "/proxy/cluster/%s"
)

var (
	Tracer = otel.Tracer("client/clusterproxy")
)

// ChangefeedClientGetter defines the interface for getting a changefeed client
type ChangefeedClientGetter interface {
	// ChangefeedClient returns a changefeed client
	ChangefeedClient() ChangefeedAPIV2
}

// ChangefeedAPIV2 defines the interface for changefeed operations through the ticdc open API v2
// note: the API payload definition is copy from ticdc api package, the data struct must be kept in sync
type ChangefeedAPIV2 interface {
	// CreateChangefeedV2 creates a new changefeed task and returns the created task information
	CreateChangefeedV2(ctx context.Context, crName string, keyspace string, model *apiv1.ChangefeedConfig) (*apiv1.Changefeed, error)
	// UpdateChangefeedV2 updates an existing paused changefeed task and returns the updated task information
	UpdateChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string, model *apiv1.ChangefeedConfig) (*apiv1.Changefeed, error)
	// GetChangefeedListV2 returns a whole list of changefeed tasks, include failed, normal, and stopped tasks
	GetChangefeedListV2(ctx context.Context, crName string, keyspace string) (*apiv1.ListChangefeedsResponse, error)
	// PauseChangefeedV2 pauses a running changefeed task
	PauseChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string) error
	// ResumeChangefeedV2 resumes a paused changefeed task with the specified new checkpoint ts
	ResumeChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string, resumeConfig *apiv1.ResumeChangefeedConfig) error
	// DeleteChangefeedV2 deletes a changefeed task
	DeleteChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string) error
	// GetChangefeedV2 returns the changefeed task information
	GetChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string) (*apiv1.Changefeed, error)
	// TestConnection tests the connection to the target downstream, or verify the changefeed configuration.
	TestConnection(ctx context.Context, req *core.CheckRequest) error
}

type changefeed struct {
	client client.Client
}

// NewChangefeedAPIV2 creates a new instance of the ChangefeedAPIV2 with the given client
func NewChangefeedAPIV2(client client.Client) ChangefeedAPIV2 {
	return &changefeed{
		client: client,
	}
}

func (p *changefeed) CreateChangefeedV2(ctx context.Context, crName string, keyspace string, req *apiv1.ChangefeedConfig) (*apiv1.Changefeed, error) {
	ctx, span := Tracer.Start(ctx, "CreateChangefeed")
	defer span.End()
	ret := &apiv1.Changefeed{}
	if err := p.client.Request(ctx, "POST", p.buildCDCApiPath(crName, cdcPathV2, keyspace),
		req, ret); err != nil {
		return nil, errors.Trace(err)
	}
	return ret, nil
}

func (p *changefeed) UpdateChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string, model *apiv1.ChangefeedConfig) (*apiv1.Changefeed, error) {
	ctx, span := Tracer.Start(ctx, "UpdateChangefeed")
	defer span.End()

	ret := &apiv1.Changefeed{}
	if err := p.client.Request(ctx, "PUT", p.buildCDCApiPath(crName, cdcPathV2+"/"+taskID, keyspace),
		model, ret); err != nil {
		return nil, errors.Trace(err)
	}
	return ret, nil
}

func (p *changefeed) GetChangefeedListV2(ctx context.Context, crName string, keyspace string) (*apiv1.ListChangefeedsResponse, error) {
	ctx, span := Tracer.Start(ctx, "GetChangefeedList")
	defer span.End()

	result := &apiv1.ListChangefeedsResponse{}
	// list all state cdc tasks, default only return normal/failed/stopped tasks
	if err := p.client.Request(ctx, "GET", p.buildCDCApiPath(crName, cdcPathV2, keyspace)+"&state=all",
		nil, result); err != nil {
		return nil, err
	}
	return result, nil
}

func (p *changefeed) PauseChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string) error {
	ctx, span := Tracer.Start(ctx, "PauseChangefeed")
	defer span.End()

	url := fmt.Sprintf("%s/%s/pause", cdcPathV2, taskID)
	return p.client.Request(ctx, "POST", p.buildCDCApiPath(crName, url, keyspace),
		nil, nil)
}

func (p *changefeed) ResumeChangefeedV2(ctx context.Context,
	crName string, keyspace string, taskID string, resumeConfig *apiv1.ResumeChangefeedConfig) error {
	ctx, span := Tracer.Start(ctx, "ResumeChangefeed")
	defer span.End()

	url := fmt.Sprintf("%s/%s/resume", cdcPathV2, taskID)
	return p.client.Request(ctx, "POST", p.buildCDCApiPath(crName, url, keyspace),
		resumeConfig, nil)
}

func (p *changefeed) DeleteChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string) error {
	ctx, span := Tracer.Start(ctx, "DeleteChangefeed")
	defer span.End()

	url := fmt.Sprintf("%s/%s", cdcPathV2, taskID)
	return p.client.Request(ctx, "DELETE", p.buildCDCApiPath(crName, url, keyspace), nil, nil)
}

func (p *changefeed) GetChangefeedV2(ctx context.Context, crName string, keyspace string, taskID string) (*apiv1.Changefeed, error) {
	ctx, span := Tracer.Start(ctx, "GetChangefeed")
	defer span.End()

	url := fmt.Sprintf("%s/%s", cdcPathV2, taskID)
	cf := &apiv1.Changefeed{}
	if err := p.client.Request(ctx, "GET", p.buildCDCApiPath(crName, url, keyspace),
		nil, cf); err != nil {
		return nil, errors.AddStack(err)
	}
	return cf, nil
}

func (p *changefeed) TestConnection(ctx context.Context, req *core.CheckRequest) error {
	ctx, span := Tracer.Start(ctx, "TestConnection")
	defer span.End()

	// Validate request
	if req == nil {
		return errors.New("CheckRequest cannot be nil")
	}
	if req.TargetType == "" {
		return errors.New("TargetType is required")
	}

	return p.client.Request(ctx, "POST", checkPath, req, nil)
}

func getBasePath(group *cspb.TiDBGroup) string {
	return fmt.Sprintf(basePath, group.GetSharedPoolId())
}

// buildCDCApiPath builds the full path for the cdc api
// the api path is something like: /ticdc/{crName}/api/v2/changefeeds?keyspace=1234
func (p *changefeed) buildCDCApiPath(crName, url string, keyspace string) string {
	if !isDefaultTiCDC(crName) {
		return fmt.Sprintf("%s%s/%s%s?keyspace=%s",
			getBasePath(p.client.GetTiDBGroup()),
			cdcRoutePath, crName, url, keyspace)
	}
	return fmt.Sprintf("%s%s%s",
		getBasePath(p.client.GetTiDBGroup()),
		cdcRoutePath, url)
}

func isDefaultTiCDC(crName string) bool {
	return crName == "" || crName == defaultTiDBName
}