package interceptor

import (
	"context"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/dataflow-service/pkg/parse"
	"go.uber.org/zap"
	"google.golang.org/grpc"
)

var clusterCheckerHttpMethodMap = map[string][]core.ClusterStatus{
	dataflow.ConnectorService_ScaleChangefeed_FullMethodName: {
		core.ClusterStatus_CLUSTER_STATUS_NORMAL,
	},
	dataflow.ConnectorService_CreateConnector_FullMethodName: {
		core.ClusterStatus_CLUSTER_STATUS_NORMAL,
	},
	dataflow.BackupService_CreateBackupSchedule_FullMethodName: {
		core.ClusterStatus_CLUSTER_STATUS_NORMAL,
	},
}

func ClusterCheckerUnaryServerInterceptor(coreCli grpcli.CoreGrpcClient) grpc.UnaryServerInterceptor {
	return func(
		ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
	) (interface{}, error) {
		ctx = log.WithTraceID(ctx)
		logger := log.GetLogger(ctx)
		if _, ok := clusterCheckerHttpMethodMap[info.FullMethod]; !ok {
			return handler(ctx, req)
		}
		url := HttpURI(ctx)
		clusterID, hasClusterID := parse.ParseClusterIDFromPath(url)
		if !hasClusterID {
			return handler(ctx, req)
		}
		cluster, bizError := coreCli.GetAnyClusterByID(ctx, clusterID)
		if bizError != nil {
			logger.Error("get cluster by id failed", zap.Error(bizError))
			return nil, bizError
		}
		statuses := clusterCheckerHttpMethodMap[info.FullMethod]
		for _, status := range statuses {
			if status == cluster.Status {
				return handler(ctx, req)
			}
		}
		return nil, bizerr_wrapper.NewInvalidInputError("cluster id=%d not in status (%v)", clusterID, statuses)
	}
}