package changefeed

import (
	"context"

	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	changefeedopenpb "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/openapi/v1beta2"
	"github.com/tidbcloud/pkgv2/bizerr"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/structpb"
)

type ConsoleChangefeedServer interface {
	ListChangefeeds(context.Context, *dataflow.ChangefeedListReq) (*dataflow.Changefeeds, error)
	GetChangefeed(context.Context, *dataflow.ChangefeedID) (*dataflow.Changefeed, error)
	GetChangefeedDetail(context.Context, *dataflow.ChangefeedID) (*structpb.Struct, error)
	ScaleChangefeed(context.Context, *dataflow.ChangefeedScaleReq) (*emptypb.Empty, error)
	ResumeChangefeed(context.Context, *dataflow.ChangefeedID) (*emptypb.Empty, error)
	PauseChangefeed(context.Context, *dataflow.ChangefeedID) (*emptypb.Empty, error)
	CreateChangefeed(context.Context, *dataflow.ChangefeedCreateReq) (*dataflow.ChangefeedID, error)
	EditChangefeed(context.Context, *dataflow.ChangefeedEditReq) (*emptypb.Empty, error)
	DeleteChangefeed(context.Context, *dataflow.ChangefeedID) (*emptypb.Empty, error)
	TestChangefeed(context.Context, *dataflow.ChangefeedCreateReq) (*emptypb.Empty, error)
}

type ConsoleChangefeedSpecServer interface {
	List(context.Context, *dataflow.ChangefeedSpecListReq) (*dataflow.ChangefeedSpecs, error)
}

type ChangefeedAdapter interface {
	ListChangefeeds(context.Context, *changefeedopenpb.ListChangefeedsRequest, uint64) (*changefeedopenpb.ListChangefeedsResponse, error)
	GetChangefeed(context.Context, *changefeedopenpb.GetChangefeedRequest) (*changefeedopenpb.Changefeed, error)
	CreateChangefeed(context.Context, *changefeedopenpb.CreateChangefeedRequest, uint64) (*changefeedopenpb.Changefeed, error)
	DeleteChangefeed(context.Context, *changefeedopenpb.DeleteChangefeedRequest) (*emptypb.Empty, error)
	PauseChangefeed(context.Context, *changefeedopenpb.PauseChangefeedRequest) (*emptypb.Empty, error)
	ResumeChangefeed(context.Context, *changefeedopenpb.ResumeChangefeedRequest) (*emptypb.Empty, error)
	EditChangefeedDownstreamConfig(context.Context, *changefeedopenpb.EditChangefeedDownstreamConfigRequest) (*changefeedopenpb.Changefeed, error)
	ScaleChangefeed(context.Context, *changefeedopenpb.ScaleChangefeedRequest) (*changefeedopenpb.Changefeed, error)
	ListCCUs(context.Context, *changefeedopenpb.ListCCUsRequest) (*changefeedopenpb.ListCCUsResponse, error)
}

var _ changefeedopenpb.ChangefeedServiceServer = (*Service)(nil)

type Service struct {
	changefeedopenpb.UnimplementedChangefeedServiceServer
	adapter ChangefeedAdapter
}

func NewService(
	consoleAPI ConsoleChangefeedServer,
	specAPI ConsoleChangefeedSpecServer,
	changefeedRepo repo.BizChangefeedRepo,
) *Service {
	return &Service{adapter: newConsoleChangefeedAdapter(consoleAPI, specAPI, changefeedRepo)}
}

func (s *Service) ListChangefeeds(
	ctx context.Context, req *changefeedopenpb.ListChangefeedsRequest,
) (*changefeedopenpb.ListChangefeedsResponse, error) {
	tidbID, err := parseUintID(req.GetTidbId(), "tidb_id")
	if err != nil {
		return nil, err
	}
	return s.adapter.ListChangefeeds(ctx, req, tidbID)
}

func (s *Service) GetChangefeed(
	ctx context.Context, req *changefeedopenpb.GetChangefeedRequest,
) (*changefeedopenpb.Changefeed, error) {
	return s.adapter.GetChangefeed(ctx, req)
}

func (s *Service) CreateChangefeed(
	ctx context.Context, req *changefeedopenpb.CreateChangefeedRequest,
) (*changefeedopenpb.Changefeed, error) {
	if req.GetChangefeed() == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}
	tidbID, err := parseUintID(req.GetChangefeed().GetTidbId(), "tidb_id")
	if err != nil {
		return nil, err
	}
	return s.adapter.CreateChangefeed(ctx, req, tidbID)
}

func (s *Service) DeleteChangefeed(
	ctx context.Context, req *changefeedopenpb.DeleteChangefeedRequest,
) (*emptypb.Empty, error) {
	return s.adapter.DeleteChangefeed(ctx, req)
}

func (s *Service) PauseChangefeed(
	ctx context.Context, req *changefeedopenpb.PauseChangefeedRequest,
) (*emptypb.Empty, error) {
	return s.adapter.PauseChangefeed(ctx, req)
}

func (s *Service) ResumeChangefeed(
	ctx context.Context, req *changefeedopenpb.ResumeChangefeedRequest,
) (*emptypb.Empty, error) {
	return s.adapter.ResumeChangefeed(ctx, req)
}

func (s *Service) EditChangefeedDownstreamConfig(
	ctx context.Context, req *changefeedopenpb.EditChangefeedDownstreamConfigRequest,
) (*changefeedopenpb.Changefeed, error) {
	return s.adapter.EditChangefeedDownstreamConfig(ctx, req)
}

func (s *Service) ScaleChangefeed(
	ctx context.Context, req *changefeedopenpb.ScaleChangefeedRequest,
) (*changefeedopenpb.Changefeed, error) {
	return s.adapter.ScaleChangefeed(ctx, req)
}

func (s *Service) ListCCUs(
	ctx context.Context, req *changefeedopenpb.ListCCUsRequest,
) (*changefeedopenpb.ListCCUsResponse, error) {
	return s.adapter.ListCCUs(ctx, req)
}