package connector

import (
	"context"

	"github.com/gogo/protobuf/types"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/model/cdc"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	"github.com/pingcap/dbaas/central/service"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	"go.uber.org/zap"
)

type connectorService struct {
	central.UnimplementedConnectorServiceServer
	service.OperationsService
	cdcModel cdc.Model
}

var _ central.ConnectorServiceServer = (*connectorService)(nil)

// NewService constructs a connectorService
func NewService(s service.OperationsService, model cdc.Model) central.ConnectorServiceServer {
	return &connectorService{OperationsService: s, cdcModel: model}
}

func (s *connectorService) userConnector(ctx context.Context) cdc.UserContext {
	user := s.CurrentUser(ctx)
	return s.cdcModel.User(user)
}

func (s *connectorService) Log(ctx context.Context) *zap.Logger {
	return s.userConnector(ctx).Log()
}

func (s *connectorService) ListConnectors(ctx context.Context, req *central.ConnectorListReq) (*central.Connectors, error) {
	context := s.userConnector(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	return context.ListCdcTask(entity.ID(req.ClusterId), req.ConnectorType)
}

func (s *connectorService) DeleteConnector(ctx context.Context, req *central.ConnectorID) (*types.Empty, error) {
	context := s.userConnector(ctx)
	logger := s.Log(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = context.DeleteCdcTask(entity.ID(req.ConnectorId), clusterID, projectID)
	if err != nil {
		logger.Error("failed to delete connector of cluster", zap.Uint64("clusterID", clusterID.Uint64()), zap.Uint64("connectorID", req.ConnectorId), zap.Error(err))
		return &types.Empty{}, errcode.NewInternalErr(errors.New("failed to delete connector"))
	}
	return &types.Empty{}, nil
}

func (s *connectorService) GetConnector(ctx context.Context, req *central.ConnectorID) (*central.Connector, error) {
	context := s.userConnector(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	return context.GetCdcTask(entity.ID(req.ConnectorId), clusterID, projectID)
}

func (s *connectorService) GetClusterTCU(ctx context.Context, req *central.GetClusterTCUReq) (*central.GetClusterTCUResp, error) {
	context := s.userConnector(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	clusterID := entity.ID(req.ClusterId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	tcuNumber, err := context.GetClusterTCU(clusterID)
	if err != nil {
		return nil, err
	}

	return &central.GetClusterTCUResp{
		TcuNumber: tcuNumber,
	}, nil
}

func (s *connectorService) CreateConnector(ctx context.Context, req *central.ConnectorCreateReq) (*central.ConnectorID, error) {
	context := s.userConnector(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}
	err = validateEndpointUrl(req.Url)
	if err != nil {
		return nil, err
	}

	return context.CreateCdcTask(
		ctx,
		entity.ID(req.ClusterId),
		projectID,
		req.ConnectorType,
		req.Url,
		req.Topic,
		req.User,
		req.Password,
		req.StartTso,
		req.IgnoreReplication)
}

func (s *connectorService) TestConnector(ctx context.Context, req *central.ConnectorCreateReq) (*types.Empty, error) {
	context := s.userConnector(ctx)
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	err := context.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	err = validateEndpointUrl(req.Url)
	if err != nil {
		return nil, err
	}

	return &types.Empty{}, context.TestCdcTask(
		ctx,
		entity.ID(req.ProjectId),
		entity.ID(req.ClusterId),
		req.ConnectorType,
		req.Url,
		req.Topic,
		req.User,
		req.Password,
		req.StartTso,
		tidbcloud.BuildClusterProxy)
}