package service

import (
	"context"
	"fmt"

	"github.com/elliotchance/pie/v2"
	infrav1 "github.com/tidbcloud/infra-api/api/v1"
	regionalpb "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	cloudv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/pkg/cloud"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	"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"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	awsPrincipalIdPattern = "arn:aws:iam::%s:root"
)

type sinkEndpointServer struct {
	apiv1.UnimplementedTiCDCSinkEndpointServiceServer
	kubeClient  client.Client
	cfg         *config.Config
	regionalCli regionalservice.Client
}

func NewSinkEndpointServiceServer(cli client.Client,
	regionalCli regionalservice.Client,
	cfg *config.Config) apiv1.TiCDCSinkEndpointServiceServer {
	return &sinkEndpointServer{
		kubeClient:  cli,
		cfg:         cfg,
		regionalCli: regionalCli,
	}
}

func (p sinkEndpointServer) CreateTiCDCSinkEndpoint(ctx context.Context, request *apiv1.CreateTiCDCSinkEndpointRequest) (*apiv1.TiCDCSinkEndpoint, error) {
	var (
		appbox *regionalpb.APPBox
		err    error
	)
	if request.TicdcSinkEndpoint.ProjectId != "" {
		appbox, err = p.regionalCli.GetAPPBoxByTenantIDAndProjectID(ctx, request.TicdcSinkEndpoint.TenantId, request.TicdcSinkEndpoint.ProjectId)
	} else {
		appbox, err = p.regionalCli.GetAPPBoxByTenantIDAndClusterID(ctx, request.TicdcSinkEndpoint.TenantId, request.TicdcSinkEndpoint.TidbClusterId)
	}
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get appbox: %s", err)
	}

	records := buildDNSRecords(request.TicdcSinkEndpoint.DnsRecords)
	endpoint := cloudv1.SinkEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name:      request.TicdcSinkEndpointId,
			Namespace: p.cfg.Namespace,
			Labels:    buildLabelsForCreatingSinkEndpoint(p.cfg.Provider, p.cfg.Region, request.TicdcSinkEndpoint.TenantId, request.TicdcSinkEndpoint.ProjectId),
		},
		Spec: cloudv1.SinkEndpointSpec{
			ServiceName:       request.TicdcSinkEndpoint.ServiceName,
			RootDomainName:    request.TicdcSinkEndpoint.RootDomainName,
			DNSRecords:        records,
			AppBoxID:          appbox.Name,
			AvailabilityZones: request.TicdcSinkEndpoint.AvailabilityZones,
		},
	}

	if err := p.kubeClient.Create(ctx, &endpoint); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "Sink endpoint %s already exists", request.TicdcSinkEndpointId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to create sink endpoint %s: %s", request.TicdcSinkEndpointId, err)
	}

	return buildAPISinkEndpointItem(endpoint), nil
}

func (p sinkEndpointServer) ListTiCDCSinkEndpoints(ctx context.Context, request *apiv1.ListTiCDCSinkEndpointsRequest) (*apiv1.ListTiCDCSinkEndpointsResponse, error) {
	list := &cloudv1.SinkEndpointList{}
	labelSet := buildLabelSetForQuery(request.ProjectId, request.TenantId, "")

	listOptions := []client.ListOption{
		client.MatchingLabelsSelector{Selector: labelSet.AsSelector()},
		client.InNamespace(p.cfg.Namespace),
	}

	if err := p.kubeClient.List(ctx, list, listOptions...); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to list sink endpoints: %s", err)
	}

	endpoints := make([]*apiv1.TiCDCSinkEndpoint, 0, len(list.Items))
	for _, endpoint := range list.Items {
		endpoints = append(endpoints, buildAPISinkEndpointItem(endpoint))
	}

	return &apiv1.ListTiCDCSinkEndpointsResponse{TicdcSinkEndpoints: endpoints}, nil
}

func (p sinkEndpointServer) GetTiCDCSinkEndpoint(ctx context.Context, request *apiv1.GetTiCDCSinkEndpointRequest) (*apiv1.TiCDCSinkEndpoint, error) {
	endpoint := cloudv1.SinkEndpoint{}
	if err := p.kubeClient.Get(ctx, client.ObjectKey{Name: request.TicdcSinkEndpointId, Namespace: p.cfg.Namespace}, &endpoint); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Sink endpoint %s not found", request.TicdcSinkEndpointId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get sink endpoint %s: %s", request.TicdcSinkEndpointId, err)
	}
	return buildAPISinkEndpointItem(endpoint), nil
}

func (p sinkEndpointServer) DeleteTiCDCSinkEndpoint(ctx context.Context, request *apiv1.DeleteTiCDCSinkEndpointRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("endpoint", request.TicdcSinkEndpointId))
	endpoint := &cloudv1.SinkEndpoint{
		ObjectMeta: metav1.ObjectMeta{
			Name:      request.TicdcSinkEndpointId,
			Namespace: p.cfg.Namespace,
		},
	}

	if err := p.kubeClient.Delete(ctx, endpoint); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Warn("Sink endpoint not found")
			return &emptypb.Empty{}, nil
		}
		return nil, status.Errorf(codes.Internal, "Failed to delete sink endpoint %s: %s", request.TicdcSinkEndpointId, err)
	}

	return &emptypb.Empty{}, nil
}

func (p sinkEndpointServer) GetSinkEndpointAuthInfo(ctx context.Context, request *apiv1.GetSinkEndpointAuthInfoRequest) (*apiv1.GetSinkEndpointAuthInfoResp, error) {
	var (
		appbox *regionalpb.APPBox
		err    error
	)
	if request.ProjectId != "" {
		appbox, err = p.regionalCli.GetAPPBoxByTenantIDAndProjectID(ctx, request.TenantId, request.ProjectId)
	} else {
		appbox, err = p.regionalCli.GetAPPBoxByTenantIDAndClusterID(ctx, request.TenantId, request.TidbClusterId)
	}
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get appbox: %s", err)
	}
	authInfo := &apiv1.GetSinkEndpointAuthInfoResp{
		Network:        appbox.Kubernetes.Vpc,
		CloudAccountId: appbox.Kubernetes.AccountId,
	}
	if p.cfg.Provider == infrav1.ProviderTypeAWS {
		authInfo.AwsPrincipalId = fmt.Sprintf(awsPrincipalIdPattern, appbox.Kubernetes.AccountId)
	}
	if request.TicdcSinkEndpointId != "" {
		endpoint := cloudv1.SinkEndpoint{}
		if err := p.kubeClient.Get(ctx, client.ObjectKey{Name: request.TicdcSinkEndpointId, Namespace: p.cfg.Namespace}, &endpoint); err != nil {
			return nil, status.Errorf(codes.Internal, "failed to get sink endpoint %s: %s", request.TicdcSinkEndpointId, err)
		}
		authInfo.VpcEndpointId = endpoint.Status.EndpointID
	}
	return authInfo, nil
}

func (p sinkEndpointServer) GetAvailableZones(ctx context.Context, request *apiv1.GetAvailableZonesRequest) (*apiv1.GetAvailableZonesResponse, error) {
	var (
		appbox *regionalpb.APPBox
		err    error
	)
	if request.ProjectId != "" {
		appbox, err = p.regionalCli.GetAPPBoxByTenantIDAndProjectID(ctx, request.TenantId, request.ProjectId)
	} else {
		appbox, err = p.regionalCli.GetAPPBoxByTenantIDAndClusterID(ctx, request.TenantId, request.TidbClusterId)
	}
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get appbox: %s", err)
	}
	var availableZones []string
	switch info := appbox.Kubernetes.Info.(type) {
	case *regionalpb.APPBox_Kubernetes_Aws_:
		availableZones = pie.Map(info.Aws.NodeSubnets, func(subnet *regionalpb.APPBox_Kubernetes_AwsSubnet) string {
			return subnet.AzId
		})
	case *regionalpb.APPBox_Kubernetes_Gcp_:
		availableZones = info.Gcp.NodeZones
	case *regionalpb.APPBox_Kubernetes_Azure_:
		availableZones = info.Azure.NodeZones
	default:
		return nil, status.Errorf(codes.Internal, "Unknown cloud provider: %T", info)
	}
	return &apiv1.GetAvailableZonesResponse{Zones: availableZones}, nil
}

func (p sinkEndpointServer) ValidateEndpointAvailabilityZones(
	ctx context.Context, request *apiv1.ValidateEndpointAvailabilityZonesRequest,
) (*emptypb.Empty, error) {
	appbox, err := p.regionalCli.GetAPPBoxByTenantIDAndProjectID(ctx, request.TenantId, request.ProjectId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get appbox: %s", err)
	}

	var cloudProvider cloud.Provider
	switch p.cfg.Provider {
	case config.ProviderAWS:
		cloudProvider = cloud.NewAWSProvider(appbox, p.cfg)
	case config.ProviderGCP:
		cloudProvider, err = cloud.NewGcpProvider(appbox, p.cfg)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "fail to build gcp Provider: %s", err)
		}
	case config.ProviderAzure:
		tid := p.cfg.Azure.CrossplaneProviderTenantID
		cid := p.cfg.Azure.CrossplaneProviderClientID
		cloudProvider, err = cloud.NewAzureProvider(ctx, tid, cid, p.regionalCli, p.kubeClient, appbox)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "fail to build azure Provider: %s", err)
		}
	default:
		return nil, status.Errorf(codes.Internal, "Unknown cloud provider: %s", p.cfg.Provider)
	}
	if err := cloudProvider.ValidateEndpointZones(ctx, request.EndpointZones, request.ServiceName); err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func buildDNSRecords(dnsRecords []*apiv1.TiCDCSinkEndpoint_DNSRecord) []cloudv1.DNSRecord {
	records := make([]cloudv1.DNSRecord, 0, len(dnsRecords))
	for _, record := range dnsRecords {
		records = append(records, cloudv1.DNSRecord{
			AvailabilityZone: record.AvailabilityZone,
			RecordName:       record.RecordName,
		})
	}
	return records
}

func buildAPISinkEndpointItem(endpoint cloudv1.SinkEndpoint) *apiv1.TiCDCSinkEndpoint {
	records := make([]*apiv1.TiCDCSinkEndpoint_DNSRecord, 0, len(endpoint.Spec.DNSRecords))
	for _, record := range endpoint.Spec.DNSRecords {
		records = append(records, &apiv1.TiCDCSinkEndpoint_DNSRecord{
			AvailabilityZone: record.AvailabilityZone,
			RecordName:       record.RecordName,
		})
	}
	return &apiv1.TiCDCSinkEndpoint{
		Id:                endpoint.Name,
		TenantId:          endpoint.Labels[infrav1.TenantTagKey],
		ProjectId:         endpoint.Labels[infrav1.ProjectTagKey],
		ServiceName:       endpoint.Spec.ServiceName,
		RootDomainName:    endpoint.Spec.RootDomainName,
		DnsRecords:        records,
		Phase:             convertSinkEndpointPhase(endpoint.Status.Phase),
		EndpointId:        endpoint.Status.EndpointID,
		EndpointAddress:   endpoint.Status.EndpointAddress,
		ZonesAddresses:    endpoint.Status.ZonesAddresses,
		AvailabilityZones: endpoint.Spec.AvailabilityZones,
		CloudAccountId:    endpoint.Status.CloudAccountID,
	}
}

func convertSinkEndpointPhase(phase cloudv1.SinkEndpointPhase) apiv1.TiCDCSinkEndpoint_Phase {
	switch phase {
	case cloudv1.EndpointResourceStateAvailable:
		return apiv1.TiCDCSinkEndpoint_AVAILABLE
	case cloudv1.EndpointResourceStatePending, cloudv1.EndpointResourceStateEndpointAvailable:
		return apiv1.TiCDCSinkEndpoint_CREATING
	case cloudv1.EndpointResourceStatePendingAcceptance:
		return apiv1.TiCDCSinkEndpoint_PENDING_ACCEPTANCE
	case cloudv1.EndpointResourceStateDeleting, cloudv1.EndpointResourceStateDeleted:
		return apiv1.TiCDCSinkEndpoint_DELETING
	case cloudv1.EndpointResourceStateRejected:
		return apiv1.TiCDCSinkEndpoint_REJECTED
	case cloudv1.EndpointResourceStateFailed, cloudv1.EndpointResourceStateExpired:
		return apiv1.TiCDCSinkEndpoint_FAILED
	default:
		return apiv1.TiCDCSinkEndpoint_PHASE_UNSPECIFIED
	}
}