package api

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy/client"
	"go.opentelemetry.io/otel"
)

const (
	cdcRoutePath   = "/ticdc"
	cdcPathV2      = "/api/v2/changefeeds"
	connectionPATH = "/test_connection"

	defaultTiDBName = "db"
)

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, 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, 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) (*apiv1.ListChangefeedsResponse, error)
	// PauseChangefeedV2 pauses a running changefeed task
	PauseChangefeedV2(ctx context.Context, crName string, taskID string) error
	// ResumeChangefeedV2 resumes a paused changefeed task with the specified new checkpoint ts
	ResumeChangefeedV2(ctx context.Context, crName string, taskID string, resumeConfig *apiv1.ResumeChangefeedConfig) error
	// DeleteChangefeedV2 deletes a changefeed task
	DeleteChangefeedV2(ctx context.Context, crName string, taskID string) error
	// GetChangefeedV2 returns the changefeed task information
	GetChangefeedV2(ctx context.Context, crName string, taskID string) (*apiv1.Changefeed, error)
	// TestConnection tests the connection to the target downstream, or verify the changefeed configuration
	TestConnection(ctx context.Context, m *ConnectionModel) 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, 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", buildCDCApiPath(crName, cdcPathV2),
		req, ret); err != nil {
		return nil, errors.Trace(err)
	}
	return ret, nil
}

func (p *changefeed) UpdateChangefeedV2(ctx context.Context, crName 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", buildCDCApiPath(crName, cdcPathV2+"/"+taskID),
		model, ret); err != nil {
		return nil, errors.Trace(err)
	}
	return ret, nil
}

func (p *changefeed) GetChangefeedListV2(ctx context.Context, crName 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
	url := cdcPathV2 + "?state=all"
	if err := p.client.Request(ctx, "GET", buildCDCApiPath(crName, url),
		nil, result); err != nil {
		return nil, err
	}
	return result, nil
}

func (p *changefeed) PauseChangefeedV2(ctx context.Context, crName 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", buildCDCApiPath(crName, url),
		nil, nil)
}

func (p *changefeed) ResumeChangefeedV2(ctx context.Context,
	crName 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", buildCDCApiPath(crName, url),
		resumeConfig, nil)
}

func (p *changefeed) DeleteChangefeedV2(ctx context.Context, crName 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", buildCDCApiPath(crName, url), nil, nil)
}

func (p *changefeed) GetChangefeedV2(ctx context.Context, crName 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", buildCDCApiPath(crName, url),
		nil, cf); err != nil {
		return nil, errors.AddStack(err)
	}
	return cf, nil
}

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

	return p.client.Request(ctx, "POST", connectionPATH, m, nil)
}

// buildCDCApiPath builds the full path for the cdc api
// the api path is something like: /ticdc/{crName}/api/v2/changefeeds
func buildCDCApiPath(crName, url string) string {
	if !isDefaultTiCDC(crName) {
		return fmt.Sprintf("%s/%s%s", cdcRoutePath, crName, url)
	}
	return fmt.Sprintf("%s%s", cdcRoutePath, url)
}

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