package service

import (
	"context"
	"fmt"

	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/ticdc-service/pkg/cloud"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	ticdc_consts "github.com/tidbcloud/ticdc-service/pkg/consts"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type certificateServer struct {
	apiv1.UnimplementedTiCDCCertificateServiceServer
	kubeClient  client.Client
	cfg         *config.Config
	regionalCli regionalservice.Client
}

func NewCertificateServer(
	cli client.Client,
	regionalCli regionalservice.Client,
	cfg *config.Config,
) apiv1.TiCDCCertificateServiceServer {
	return &certificateServer{
		kubeClient:  cli,
		cfg:         cfg,
		regionalCli: regionalCli,
	}
}

func (s *certificateServer) CreateCertificate(ctx context.Context, req *apiv1.CreateCertificateRequest) (*apiv1.Certificate, error) {
	certificate := req.Certificate
	certificate.Id = req.CertificateId
	if len(certificate.Items) == 0 {
		return nil, status.Error(codes.InvalidArgument, "certificate items can not be empty")
	}

	secret := s.buildDataPlaneSecret(req.TidbClusterId, certificate)
	dataPlaneCli, err := s.getDataPlaneClient(ctx, req.TidbClusterId)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	if err := dataPlaneCli.Create(ctx, &secret); err != nil {
		if apierrors.IsAlreadyExists(err) {
			tilog.WithContextL(ctx).Info("secret already exists", zap.String("certificate_id", req.CertificateId))
			return certificate, nil
		}
		return nil, status.Errorf(codes.Internal, "create secret error: %s", err.Error())
	}
	return certificate, nil
}

func (s *certificateServer) GetCertificate(ctx context.Context, req *apiv1.GetCertificateRequest) (*apiv1.Certificate, error) {
	dataPlaneCli, err := s.getDataPlaneClient(ctx, req.TidbClusterId)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.CertificateId,
			Namespace: s.getSecretNamespace(req.TidbClusterId),
		},
	}
	if err := dataPlaneCli.Get(ctx, client.ObjectKeyFromObject(secret), secret); err != nil {
		return nil, status.Errorf(codes.Internal, "get secret error: %s", err.Error())
	}

	return s.extractCertificateFromSecret(secret), nil
}

func (s *certificateServer) DeleteCertificate(ctx context.Context, req *apiv1.DeleteCertificateRequest) (*emptypb.Empty, error) {
	dataPlaneCli, err := s.getDataPlaneClient(ctx, req.TidbClusterId)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.CertificateId,
			Namespace: s.getSecretNamespace(req.TidbClusterId),
		},
	}
	if err := dataPlaneCli.Delete(ctx, secret); err != nil {
		return nil, status.Errorf(codes.Internal, "delete secret error: %s", err.Error())
	}
	return &emptypb.Empty{}, nil
}

func (s *certificateServer) buildDataPlaneSecret(tidbClusterID string, certificate *apiv1.Certificate) corev1.Secret {
	secret := corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      certificate.Id,
			Namespace: s.getSecretNamespace(tidbClusterID),
			Labels: map[string]string{
				ticdc_consts.TiCDCCertLabel: "true",
			},
		},
		StringData: make(map[string]string, len(certificate.Items)),
		Type:       corev1.SecretTypeOpaque,
	}
	for _, item := range certificate.Items {
		secret.StringData[item.Name] = item.Content
	}
	return secret
}

func (s *certificateServer) extractCertificateFromSecret(secret *corev1.Secret) *apiv1.Certificate {
	certificate := &apiv1.Certificate{
		Id:    secret.Name,
		Items: make([]*apiv1.Certificate_Item, 0, len(secret.Data)),
	}
	for k, v := range secret.Data {
		certificate.Items = append(certificate.Items, &apiv1.Certificate_Item{
			Name:    k,
			Content: string(v),
		})
	}
	return certificate
}

func (s *certificateServer) getDataPlaneClient(ctx context.Context, clusterID string) (client.Client, error) {
	cluster, err := s.regionalCli.GetCluster(ctx, clusterID)
	if err != nil {
		return nil, fmt.Errorf("failed to get cluster %s: %w", clusterID, err)
	}
	appBox, err := s.regionalCli.GetAPPBoxByTenantIDAndProjectID(ctx, cluster.TenantId, cluster.ProjectId)
	if err != nil {
		return nil, fmt.Errorf("failed to get appbox: %w", err)
	}

	var provider cloud.Provider
	switch s.cfg.Provider {
	case config.ProviderAWS:
		provider = cloud.NewAWSProvider(appBox, s.cfg)
	case config.ProviderGCP:
		provider, err = cloud.NewGcpProvider(appBox, s.cfg)
	case config.ProviderAzure:
		tid := s.cfg.Azure.CrossplaneProviderTenantID
		cid := s.cfg.Azure.CrossplaneProviderClientID
		provider, err = cloud.NewAzureProvider(ctx, tid, cid, s.regionalCli, s.kubeClient, appBox)
	default:
		return nil, fmt.Errorf("unsupported provider: %s", s.cfg.Provider)
	}
	if err != nil {
		return nil, fmt.Errorf("failed to build cloud provider: %w", err)
	}

	dataPlaneCli, err := provider.GetKubeClient(ctx)
	if err != nil {
		return nil, fmt.Errorf("failed to get data plane client: %w", err)
	}
	return dataPlaneCli, nil
}

func (s *certificateServer) getSecretNamespace(tidbClusterID string) string {
	return fmt.Sprintf("tidb%s", tidbClusterID)
}