package changefeed

import (
	"context"

	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1"
	"github.com/tidbcloud/dataflow-service/internal/openapi/service"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/tiinf/sdk/auth"
	"google.golang.org/protobuf/types/known/emptypb"
)

type RBACClient interface {
	CheckIdentityPermission(context.Context, *auth.CheckPermissionParam) (*auth.CheckPermissionRet, error)
}

type RBACClientFunc func(context.Context, *auth.CheckPermissionParam) (*auth.CheckPermissionRet, error)

func (f RBACClientFunc) CheckIdentityPermission(ctx context.Context,
	params *auth.CheckPermissionParam,
) (*auth.CheckPermissionRet, error) {
	return f(ctx, params)
}

type ConnectorServiceServer interface {
	CreateConnector(context.Context, *dataflow.ConnectorCreateReq) (*dataflow.ConnectorID, error)
	DeleteConnector(context.Context, *dataflow.ConnectorID) (*emptypb.Empty, error)
	ScaleChangefeed(context.Context, *dataflow.ChangefeedScaleReq) (*emptypb.Empty, error)
	TestConnector(context.Context, *dataflow.ConnectorCreateReq) (*emptypb.Empty, error)
	ListConnectors(context.Context, *dataflow.ConnectorListReq) (*dataflow.Connectors, error)
	GetConnector(context.Context, *dataflow.ConnectorID) (*dataflow.Connector, error)
	StartConnector(context.Context, *dataflow.ConnectorID) (*emptypb.Empty, error)
	StopConnector(context.Context, *dataflow.ConnectorID) (*emptypb.Empty, error)
	EditConnector(context.Context, *dataflow.ConnectorEditReq) (*emptypb.Empty, error)
}

type ChangefeedAdapter interface {
	ResolveChangefeedScope(ctx context.Context, changefeedID string) (changefeedScope, error)
	ListChangefeeds(ctx context.Context, request *changefeedopenbp.ListChangefeedsRequest, scope changefeedScope) (*changefeedopenbp.ListChangefeedsResponse, error)
	CreateChangefeed(ctx context.Context, request *changefeedopenbp.CreateChangefeedRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error)
	DeleteChangefeed(ctx context.Context, request *changefeedopenbp.DeleteChangefeedRequest, scope changefeedScope) (*emptypb.Empty, error)
	EditChangefeedDownstreamConfig(ctx context.Context, request *changefeedopenbp.EditChangefeedDownstreamConfigRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error)
	GetChangefeed(ctx context.Context, request *changefeedopenbp.GetChangefeedRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error)
	GetChangefeedKernelInfo(ctx context.Context, request *changefeedopenbp.GetChangefeedKernelInfoRequest) (*changefeedopenbp.ChangefeedKernelInfo, error)
	PauseChangefeed(ctx context.Context, request *changefeedopenbp.PauseChangefeedRequest, scope changefeedScope) (*emptypb.Empty, error)
	ListChangefeedRCUs(ctx context.Context, request *changefeedopenbp.ListChangefeedRCUsRequest) (*changefeedopenbp.ListChangefeedRCUsResponse, error)
	ScaleChangefeed(ctx context.Context, request *changefeedopenbp.ScaleChangefeedRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error)
	ResumeChangefeed(ctx context.Context, request *changefeedopenbp.ResumeChangefeedRequest, scope changefeedScope) (*emptypb.Empty, error)
	GetEventFilters(ctx context.Context, request *changefeedopenbp.GetEventFiltersRequest, scope changefeedScope) (*changefeedopenbp.EventFilters, error)
	SetEventFilters(ctx context.Context, request *changefeedopenbp.SetEventFiltersRequest, scope changefeedScope) (*changefeedopenbp.SetEventFiltersResponse, error)
}

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

type Service struct {
	changefeedopenbp.UnimplementedChangefeedServiceServer
	service.BaseService

	adapter     ChangefeedAdapter
	mgmtCli     grpcli.MgmtSvcGrpcClient
	rbacChecker *rbacChecker
}

func NewService(connectorAPI ConnectorServiceServer,
	instanceTypeModel ticdc_spec.CDCInstanceTypeModel,
	changefeedModel biz_changefeed.BizChangefeedModel,
	bizChangefeedRepo repo.BizChangefeedRepo,
	ticdcClusterRepo repo.TiCDCClusterRepo,
	cdcClusterConfigManager config_manager.TiCDCClusterConfigManager,
	client RBACClient,
	mgmtCli grpcli.MgmtSvcGrpcClient,
) *Service {
	return &Service{
		adapter: newConnectorChangefeedAdapter(
			connectorAPI, instanceTypeModel, changefeedModel, bizChangefeedRepo, ticdcClusterRepo, cdcClusterConfigManager),
		rbacChecker: nwRBACChecker(client),
		mgmtCli:     mgmtCli,
	}
}

func (s *Service) ListChangefeeds(ctx context.Context,
	request *changefeedopenbp.ListChangefeedsRequest,
) (*changefeedopenbp.ListChangefeedsResponse, error) {
	scope, err := s.authorizedClusterScope(ctx, request.GetClusterId(), ListChangefeedsReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.ListChangefeeds(ctx, request, scope)
}

func (s *Service) GetChangefeedKernelInfo(ctx context.Context,
	request *changefeedopenbp.GetChangefeedKernelInfoRequest,
) (*changefeedopenbp.ChangefeedKernelInfo, error) {
	// Internal-only API. The changefeed get permission is reused: the endpoint
	// only reads kernel metadata of a changefeed the caller can already read,
	// so no separate RBAC resource is registered for it.
	_, err := s.authorizedChangefeedScope(ctx, request.GetChangefeedId(), GetChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.GetChangefeedKernelInfo(ctx, request)
}

func (s *Service) CreateChangefeed(ctx context.Context,
	req *changefeedopenbp.CreateChangefeedRequest,
) (*changefeedopenbp.Changefeed, error) {
	if req.GetChangefeed() == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "changefeed is required")
	}
	scope, err := s.authorizedClusterScope(ctx, req.GetChangefeed().GetClusterId(), CreateChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.CreateChangefeed(ctx, req, scope)
}

func (s *Service) GetChangefeed(ctx context.Context,
	req *changefeedopenbp.GetChangefeedRequest,
) (*changefeedopenbp.Changefeed, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), GetChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.GetChangefeed(ctx, req, scope)
}

func (s *Service) PauseChangefeed(ctx context.Context,
	req *changefeedopenbp.PauseChangefeedRequest,
) (*emptypb.Empty, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), PauseChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.PauseChangefeed(ctx, req, scope)
}

func (s *Service) ResumeChangefeed(ctx context.Context,
	req *changefeedopenbp.ResumeChangefeedRequest,
) (*emptypb.Empty, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), ResumeChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.ResumeChangefeed(ctx, req, scope)
}

func (s *Service) GetEventFilters(ctx context.Context,
	req *changefeedopenbp.GetEventFiltersRequest,
) (*changefeedopenbp.EventFilters, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), GetChangefeedFiltersReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.GetEventFilters(ctx, req, scope)
}

func (s *Service) SetEventFilters(ctx context.Context,
	req *changefeedopenbp.SetEventFiltersRequest,
) (*changefeedopenbp.SetEventFiltersResponse, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), SetChangefeedFiltersReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.SetEventFilters(ctx, req, scope)
}

func (s *Service) DeleteChangefeed(ctx context.Context,
	req *changefeedopenbp.DeleteChangefeedRequest,
) (*emptypb.Empty, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), DeleteChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.DeleteChangefeed(ctx, req, scope)
}

func (s *Service) EditChangefeedDownstreamConfig(ctx context.Context,
	req *changefeedopenbp.EditChangefeedDownstreamConfigRequest,
) (*changefeedopenbp.Changefeed, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), EditChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.EditChangefeedDownstreamConfig(ctx, req, scope)
}

func (s *Service) ScaleChangefeed(ctx context.Context,
	req *changefeedopenbp.ScaleChangefeedRequest,
) (*changefeedopenbp.Changefeed, error) {
	scope, err := s.authorizedChangefeedScope(ctx, req.GetChangefeedId(), ScaleChangefeedReq)
	if err != nil {
		return nil, err
	}
	return s.adapter.ScaleChangefeed(ctx, req, scope)
}

func (s *Service) ListChangefeedRCUs(ctx context.Context,
	req *changefeedopenbp.ListChangefeedRCUsRequest,
) (*changefeedopenbp.ListChangefeedRCUsResponse, error) {
	return s.adapter.ListChangefeedRCUs(ctx, req)
}

func (s *Service) authorizedChangefeedScope(ctx context.Context,
	changefeedID string, requestURI string,
) (changefeedScope, error) {
	scope, err := s.adapter.ResolveChangefeedScope(ctx, changefeedID)
	if err != nil {
		return changefeedScope{}, err
	}
	if err := s.rbacChecker.checkPermission(ctx, requestURI,
		scope.tenantID(), scope.projectID(), scope.clusterID()); err != nil {
		return changefeedScope{}, err
	}
	return scope, nil
}

func (s *Service) authorizedClusterScope(ctx context.Context,
	clusterID string, requestURI string,
) (changefeedScope, error) {
	if clusterID == "" {
		return changefeedScope{}, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}
	clusterIDNum, err := parseUintID(clusterID, "cluster_id")
	if err != nil {
		return changefeedScope{}, err
	}

	cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, clusterIDNum)
	if err != nil {
		return changefeedScope{}, err
	}

	scope := newChangefeedScope(cluster)
	if err := s.rbacChecker.checkPermission(ctx, requestURI,
		scope.tenantID(), scope.projectID(), scope.clusterID()); err != nil {
		return changefeedScope{}, err
	}
	return scope, nil
}