package managerserver

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	pbm "github.com/tidbcloud/aws-shared-provider/pbgen/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/notifier"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbmanager"
	ep "github.com/tidbcloud/aws-shared-provider/pkg/utils/endpoint"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/random"
)

type grpcServer struct {
	ctx context.Context

	tManagers map[string]tidbmanager.TidbManager
	notifier  *notifier.Notifier
	pbm.UnimplementedTidbBackendServer
}

func (s *grpcServer) WatchBackendV2(
	request *pbm.WatchBackendRequestV2,
	stream pbm.TidbBackend_WatchBackendV2Server,
) error {
	logger := log.WithName("WatchBackendV2")
	if request.Token == "" {
		logger.Error(noTokenErr, "get token")
		return noTokenErr
	}

	notifierID := fmt.Sprintf(
		"%s-%s-%s", request.Token, request.GatewayHostname, random.Random(6), //nolint: gomnd // not need
	)
	logger = logger.WithValues("token", request.Token).WithValues("notifierID", notifierID)
	logger.V(1).Info("grpc req param", "req", request)

	endpointsChan := s.notifier.Register(request.Token, notifierID)
	defer s.notifier.Unregister(request.Token, notifierID)

	for tidbPool, tManager := range s.tManagers {
		if endpoints := tManager.GetAssignedTidbByToken(request.Token); endpoints != nil {
			backends := make([]*pbm.Backend, 0, len(endpoints.Endpoints))
			for _, endpoint := range endpoints.Endpoints {
				podName, podIP := ep.ParsePodFromEndpoint(endpoint)
				backends = append(backends, &pbm.Backend{PodName: podName, Ip: podIP})
			}
			br := &pbm.BackendResponseV2{
				Backends:    backends,
				TidbPool:    tidbPool,
				Comment:     "init assign tidb backend",
				ModRevision: endpoints.ModRevision,
				Replicas:    int32(endpoints.Replicas),
			}
			if err := stream.Send(br); err != nil {
				logger.Error(err, "grpc send init assign tidb backend error", "br", br)
				return err
			}
			logger.V(1).Info("grpc send init assign tidb backend", "br", br)
		}
	}

	for {
		select {
		case <-s.ctx.Done():
			return s.ctx.Err()
		case <-stream.Context().Done():
			err := stream.Context().Err()
			logger.Info("grpc stream ctx done", "err", err)
			return err
		case endpoints, ok := <-endpointsChan:
			if !ok {
				err := errors.New("the endpoints channel closed, may be it's full")
				logger.Error(err, "the connection will be disconnected")
				return err
			}
			eds := endpoints.(*ep.Endpoints)
			backends := make([]*pbm.Backend, 0, len(eds.Endpoints))
			for _, endpoint := range eds.Endpoints {
				podName, podIP := ep.ParsePodFromEndpoint(endpoint)
				backends = append(backends, &pbm.Backend{PodName: podName, Ip: podIP})
			}
			br := &pbm.BackendResponseV2{
				Backends:    backends,
				TidbPool:    eds.TidbPool,
				Comment:     eds.Comment,
				ModRevision: eds.ModRevision,
				Replicas:    int32(eds.Replicas),
			}
			if err := stream.Send(br); err != nil {
				logger.Error(err, "grpc send backend error", "br", br)
				return err
			}
			logger.V(1).Info("grpc send backend", "br", br)
		}
	}
}

func (s *grpcServer) GetBackendV2(
	ctx context.Context,
	request *pbm.GetBackendRequestV2,
) (*pbm.GetBackendResponseV2, error) {
	logger := log.WithName("GetBackendV2")
	if request.Token == "" {
		logger.Error(noTokenErr, "get token")
		return nil, noTokenErr
	}

	responses := make([]*pbm.BackendResponseV2, 0)
	for tidbPool, tManager := range s.tManagers {
		endpoints := tManager.GetAssignedTidbByToken(request.Token)
		if endpoints == nil {
			continue
		}
		backends := make([]*pbm.Backend, 0, len(endpoints.Endpoints))
		for _, endpoint := range endpoints.Endpoints {
			podName, podIP := ep.ParsePodFromEndpoint(endpoint)
			backends = append(backends, &pbm.Backend{PodName: podName, Ip: podIP})
		}
		responses = append(responses, &pbm.BackendResponseV2{
			Backends:    backends,
			TidbPool:    tidbPool,
			Comment:     endpoints.Comment,
			ModRevision: endpoints.ModRevision,
			Replicas:    int32(endpoints.Replicas),
		})
	}
	return &pbm.GetBackendResponseV2{Responses: responses}, nil
}