package service

import (
	"context"
	"errors"
	"fmt"
	"strings"

	regionalpb "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/ticdc-service-ng/pkg/config"
	"github.com/tidbcloud/ticdc-service-ng/pkg/regionalservice"
	"github.com/tidbcloud/ticdc-service-ng/pkg/util"
)

const (
	awsPrincipalIdPattern = "arn:aws:iam::%s:root"
	tenantTag             = "tags.tidbcloud.com/tenant"
	sharedPoolTag         = "tags.tidbcloud.com/shared-pool"
	componentTag          = "tags.tidbcloud.com/component"
)

type sinkEndpointServer struct {
	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) {
	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, request.TicdcSinkEndpoint.SharedPoolId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get shared pool: %s", err)
	}
	dataplane, err := p.regionalCli.GetDataPlaneByID(ctx, sharedPool.DataplaneName)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get data plane: %s", err)
	}
	regionalEndpoint := &regionalpb.PrivateLinkEndpoint{
		Id: request.TicdcSinkEndpointId,
		Tags: map[string]string{
			tenantTag:     request.TicdcSinkEndpoint.TenantId,
			componentTag:  "ticdc",
			sharedPoolTag: request.TicdcSinkEndpoint.SharedPoolId,
		},
		RootDomainName: request.TicdcSinkEndpoint.RootDomainName,
	}
	var dnsRecords = make([]*regionalpb.PrivateLinkEndpoint_DNSRecord, 0, len(request.TicdcSinkEndpoint.DnsRecords))
	for _, record := range request.TicdcSinkEndpoint.DnsRecords {
		dnsRecords = append(dnsRecords, &regionalpb.PrivateLinkEndpoint_DNSRecord{
			AvailabilityZone: record.AvailabilityZone,
			RecordName:       record.RecordName,
		})
	}

	// Amazon MSK multi-VPC connectivity: only the MSK ARN + auth scheme are
	// passed down; the customer endpoint service / DNS records do not apply.
	if request.TicdcSinkEndpoint.EndpointType == apiv1.TiCDCSinkEndpoint_AMAZON_MSK {
		if p.cfg.Provider != config.ProviderAWS {
			return nil, status.Error(codes.InvalidArgument, "amazon msk sink endpoint is only supported on AWS")
		}
		if request.TicdcSinkEndpoint.MskArn == "" {
			return nil, status.Error(codes.InvalidArgument, "msk_arn is required for amazon msk sink endpoint")
		}
		// Only SASL/SCRAM is implemented today; reject unsupported / unspecified
		// auth types here so callers get InvalidArgument instead of an Internal
		// error surfaced from dataplane-manager.
		if request.TicdcSinkEndpoint.AuthType != apiv1.TiCDCSinkEndpoint_SASL_SCRAM {
			return nil, status.Errorf(codes.InvalidArgument,
				"amazon msk auth type %s is not supported", request.TicdcSinkEndpoint.AuthType)
		}
		regionalEndpoint.Info = &regionalpb.PrivateLinkEndpoint_AwsMsk_{
			AwsMsk: &regionalpb.PrivateLinkEndpoint_AwsMsk{
				MskArn:   request.TicdcSinkEndpoint.MskArn,
				AuthType: mskAuthTypeToRegional(request.TicdcSinkEndpoint.AuthType),
			},
		}
		endpoint, err := p.regionalCli.CreateSinkEndpoint(ctx, &regionalpb.CreatePrivateLinkEndpointRequest{
			DataPlaneId:         sharedPool.DataplaneName,
			Id:                  request.TicdcSinkEndpointId,
			PrivateLinkEndpoint: regionalEndpoint,
		})
		if err != nil {
			return nil, status.Errorf(codes.Internal, "Failed to create tiCDCSink endpoint: %s", err)
		}
		return assembleApiEndpoint(dataplane.AccountId, sharedPool.Id, endpoint), nil
	}

	switch p.cfg.Provider {
	case config.ProviderAWS:
		regionalEndpoint.Info = &regionalpb.PrivateLinkEndpoint_Aws_{
			Aws: &regionalpb.PrivateLinkEndpoint_Aws{
				EndpointServiceName: request.TicdcSinkEndpoint.ServiceName,
				DnsRecords:          dnsRecords,
				ServiceRegion:       p.cfg.Region,
				AzIds:               request.TicdcSinkEndpoint.AvailabilityZones,
			}}
	case config.ProviderAliCloud:
		regionalEndpoint.Info = &regionalpb.PrivateLinkEndpoint_AliCloud_{
			AliCloud: &regionalpb.PrivateLinkEndpoint_AliCloud{
				EndpointServiceName: request.TicdcSinkEndpoint.ServiceName,
				DnsRecords:          dnsRecords,
				AzIds:               request.TicdcSinkEndpoint.AvailabilityZones,
			}}
	case config.ProviderAzure:
		regionalEndpoint.Info = &regionalpb.PrivateLinkEndpoint_Azure_{
			Azure: &regionalpb.PrivateLinkEndpoint_Azure{
				EndpointServiceAlias: request.TicdcSinkEndpoint.ServiceName,
			}}
	default:
		return nil, status.Error(codes.InvalidArgument, "provider not supported")
	}
	endpoint, err := p.regionalCli.CreateSinkEndpoint(ctx, &regionalpb.CreatePrivateLinkEndpointRequest{
		DataPlaneId:         sharedPool.DataplaneName,
		Id:                  request.TicdcSinkEndpointId,
		PrivateLinkEndpoint: regionalEndpoint,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to create tiCDCSink endpoint: %s", err)
	}
	return assembleApiEndpoint(dataplane.AccountId, sharedPool.Id, endpoint), nil
}

func (p *sinkEndpointServer) ListTiCDCSinkEndpoints(ctx context.Context, request *apiv1.ListTiCDCSinkEndpointsRequest) (*apiv1.ListTiCDCSinkEndpointsResponse, error) {
	dataplanes, err := p.regionalCli.ListDataPlane(ctx)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to list data plane: %s", err)
	}
	ret := &apiv1.ListTiCDCSinkEndpointsResponse{}
	for _, dataplane := range dataplanes.DataPlanes {
		if request.TenantId != "" && dataplane.Tags[tenantTag] != request.TenantId {
			continue
		}
		sharedPoolID := dataplane.Tags[sharedPoolTag]
		if request.SharedPoolId != "" && sharedPoolID != request.SharedPoolId {
			continue
		}
		resp, err := p.regionalCli.ListPrivateLinkEndpoint(ctx, &regionalpb.ListPrivateLinkEndpointsRequest{
			DataPlaneId: dataplane.Id,
		})
		if err != nil {
			return nil, status.Errorf(codes.Internal, "Failed to list tiCDCSink endpoints: %s", err)
		}
		for _, endpoint := range resp.PrivateLinkEndpoints {
			endpointInfo := assembleApiEndpoint(dataplane.AccountId, sharedPoolID, endpoint)
			ret.TicdcSinkEndpoints = append(ret.TicdcSinkEndpoints, endpointInfo)
		}
	}
	return ret, nil
}

func (p *sinkEndpointServer) GetTiCDCSinkEndpoint(ctx context.Context, request *apiv1.GetTiCDCSinkEndpointRequest) (*apiv1.TiCDCSinkEndpoint, error) {
	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, request.GetSharedPoolId())
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get shared pool: %s", err)
	}
	dataplane, err := p.regionalCli.GetDataPlaneByID(ctx, sharedPool.DataplaneName)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get data plane: %s", err)
	}
	endpoint, err := p.regionalCli.GetPrivateLinkEndpoint(ctx, &regionalpb.GetPrivateLinkEndpointRequest{
		DataPlaneId: sharedPool.DataplaneName,
		Id:          request.TicdcSinkEndpointId,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get dataplane: %s", err)
	}
	return assembleApiEndpoint(dataplane.AccountId, sharedPool.Id, endpoint), nil
}

func (p *sinkEndpointServer) DeleteTiCDCSinkEndpoint(ctx context.Context, request *apiv1.DeleteTiCDCSinkEndpointRequest) (*emptypb.Empty, error) {
	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, request.SharedPoolId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get shared pool: %s", err)
	}
	err = p.regionalCli.DeletePrivateLinkEndpoint(ctx, &regionalpb.DeletePrivateLinkEndpointRequest{
		DataPlaneId: sharedPool.DataplaneName,
		Id:          request.TicdcSinkEndpointId,
	})
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to delete sinkendpoint: %s", err)
	}
	return &emptypb.Empty{}, nil
}

func (p *sinkEndpointServer) GetSinkEndpointAuthInfo(ctx context.Context, request *apiv1.GetSinkEndpointAuthInfoRequest) (*apiv1.GetSinkEndpointAuthInfoResp, error) {
	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, request.SharedPoolId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get shared pool: %s", err)
	}
	dataplane, err := p.regionalCli.GetDataPlaneByID(ctx, sharedPool.DataplaneName)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get dataplane: %s", err)
	}
	return &apiv1.GetSinkEndpointAuthInfoResp{
		CloudAccountId: dataplane.AccountId,
		AwsPrincipalId: fmt.Sprintf(awsPrincipalIdPattern, dataplane.AccountId),
	}, nil
}

func (p *sinkEndpointServer) GetAvailableZones(ctx context.Context, request *apiv1.GetAvailableZonesRequest) (*apiv1.GetAvailableZonesResponse, error) {
	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, request.SharedPoolId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get shared pool: %s", err)
	}
	dataPlane, err := p.regionalCli.GetDataPlaneByID(ctx, sharedPool.DataplaneName)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get dataplane: %s", err)
	}
	return &apiv1.GetAvailableZonesResponse{Zones: dataPlane.Vpc.AvailabilityZoneIds}, nil
}

func (p *sinkEndpointServer) DescribeEndpointServiceAvailabilityZones(ctx context.Context, request *apiv1.DescribeEndpointServiceAvailabilityZonesRequest) (*apiv1.DescribeEndpointServiceAvailabilityZonesResponse, error) {
	if request == nil {
		return nil, status.Error(codes.InvalidArgument, "request is required")
	}
	if p.cfg.Provider != config.ProviderAWS && p.cfg.Provider != config.ProviderAliCloud {
		return nil, status.Errorf(codes.InvalidArgument, "provider %s does not support endpoint service availability zone discovery", p.cfg.Provider)
	}

	sharedPoolID := strings.TrimSpace(request.SharedPoolId)
	serviceName := strings.TrimSpace(request.ServiceName)
	serviceRegion := strings.TrimSpace(request.ServiceRegion)
	if sharedPoolID == "" || serviceName == "" || serviceRegion == "" {
		return nil, status.Error(codes.InvalidArgument, "shared_pool_id, service_name and service_region are required")
	}
	currentRegion := strings.TrimSpace(p.cfg.Region)
	if currentRegion == "" {
		return nil, status.Error(codes.Internal, "service region is not configured")
	}
	if serviceRegion != currentRegion {
		return nil, status.Errorf(codes.InvalidArgument, "cross-region endpoint service availability zone discovery is not supported: service_region must be %s", currentRegion)
	}

	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, sharedPoolID)
	if err != nil {
		return nil, mapEndpointServiceAvailabilityZonesError("get shared pool", err)
	}
	resp, err := p.regionalCli.DescribePrivateLinkEndpointServiceZones(ctx, &regionalpb.DescribePrivateLinkEndpointServiceZonesRequest{
		DataPlaneId:   sharedPool.DataplaneName,
		ServiceName:   serviceName,
		ServiceRegion: serviceRegion,
	})
	if err != nil {
		return nil, mapEndpointServiceAvailabilityZonesError("describe endpoint service availability zones", err)
	}

	zones := make([]*apiv1.EndpointServiceAvailabilityZone, 0, len(resp.Zones))
	for _, zone := range resp.Zones {
		zones = append(zones, &apiv1.EndpointServiceAvailabilityZone{
			AvailabilityZone:   zone.AvailabilityZone,
			AvailabilityZoneId: zone.AvailabilityZoneId,
		})
	}
	return &apiv1.DescribeEndpointServiceAvailabilityZonesResponse{Zones: zones}, nil
}

func mapEndpointServiceAvailabilityZonesError(op string, err error) error {
	if errors.Is(err, context.Canceled) {
		return status.Errorf(codes.Canceled, "%s canceled", op)
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return status.Errorf(codes.DeadlineExceeded, "%s timed out", op)
	}

	var httpErr *util.StatusError
	if errors.As(err, &httpErr) {
		return status.Errorf(grpcCodeFromHTTPStatus(httpErr.StatusCode), "%s failed: status=%d", op, httpErr.StatusCode)
	}

	return status.Errorf(codes.Internal, "%s failed: %s", op, err)
}

func (p *sinkEndpointServer) ValidateEndpointAvailabilityZones(ctx context.Context, request *apiv1.ValidateEndpointAvailabilityZonesRequest) (*emptypb.Empty, error) {
	sharedPool, err := p.regionalCli.GetSharedPoolByID(ctx, request.SharedPoolId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get shared pool: %s", err)
	}
	if p.cfg.Provider == config.ProviderAzure {
		return &emptypb.Empty{}, nil
	}
	return &emptypb.Empty{}, p.regionalCli.ValidatePrivateLinkEndpointZones(ctx, &regionalpb.ValidatePrivateLinkEndpointZonesRequest{
		DataPlaneId: sharedPool.DataplaneName,
		ServiceName: request.ServiceName,
		Zones:       request.EndpointZones,
	})
}

func assembleApiEndpoint(cloudAccountID string, sharedPoolId string,
	endpoint *regionalpb.PrivateLinkEndpoint) *apiv1.TiCDCSinkEndpoint {
	ret := &apiv1.TiCDCSinkEndpoint{
		Id:             endpoint.Id,
		SharedPoolId:   sharedPoolId,
		RootDomainName: endpoint.RootDomainName,
		Phase:          convertSinkEndpointPhase(endpoint.Status),
		CloudAccountId: cloudAccountID,
		// Default to the endpoint-service model; the AwsMsk case below overrides it.
		EndpointType: apiv1.TiCDCSinkEndpoint_VPC_ENDPOINT_SERVICE,
	}
	switch info := endpoint.Info.(type) {
	case *regionalpb.PrivateLinkEndpoint_Aws_:
		awsInfo := info.Aws
		ret.ServiceName = info.Aws.EndpointServiceName
		ret.RootDomainName = endpoint.RootDomainName
		ret.DnsRecords = convertDNSRecords(awsInfo.DnsRecords)
		ret.EndpointId = awsInfo.EndpointId
		ret.EndpointAddress = awsInfo.EndpointAddress
	case *regionalpb.PrivateLinkEndpoint_AliCloud_:
		aliInfo := info.AliCloud
		ret.ServiceName = aliInfo.EndpointServiceName
		ret.DnsRecords = convertDNSRecords(aliInfo.DnsRecords)
		ret.EndpointId = aliInfo.EndpointId
		ret.EndpointAddress = aliInfo.EndpointAddress
	case *regionalpb.PrivateLinkEndpoint_Azure_:
		azureInfo := info.Azure
		ret.ServiceName = azureInfo.EndpointServiceAlias
		ret.EndpointId = azureInfo.EndpointId
		ret.EndpointAddress = azureInfo.EndpointAddress
	case *regionalpb.PrivateLinkEndpoint_AwsMsk_:
		mskInfo := info.AwsMsk
		ret.EndpointType = apiv1.TiCDCSinkEndpoint_AMAZON_MSK
		ret.MskArn = mskInfo.MskArn
		ret.AuthType = mskAuthTypeFromRegional(mskInfo.AuthType)
		ret.BootstrapBroker = mskInfo.BootstrapBroker
	}
	return ret
}

func mskAuthTypeToRegional(t apiv1.TiCDCSinkEndpoint_MskAuthType) regionalpb.PrivateLinkEndpoint_AuthType {
	switch t {
	case apiv1.TiCDCSinkEndpoint_SASL_SCRAM:
		return regionalpb.PrivateLinkEndpoint_SASL_SCRAM
	case apiv1.TiCDCSinkEndpoint_SASL_IAM:
		return regionalpb.PrivateLinkEndpoint_SASL_IAM
	default:
		return regionalpb.PrivateLinkEndpoint_AUTH_TYPE_UNSPECIFIED
	}
}

func mskAuthTypeFromRegional(t regionalpb.PrivateLinkEndpoint_AuthType) apiv1.TiCDCSinkEndpoint_MskAuthType {
	switch t {
	case regionalpb.PrivateLinkEndpoint_SASL_SCRAM:
		return apiv1.TiCDCSinkEndpoint_SASL_SCRAM
	case regionalpb.PrivateLinkEndpoint_SASL_IAM:
		return apiv1.TiCDCSinkEndpoint_SASL_IAM
	default:
		return apiv1.TiCDCSinkEndpoint_MSK_AUTH_TYPE_UNSPECIFIED
	}
}

func convertDNSRecords(records []*regionalpb.PrivateLinkEndpoint_DNSRecord) []*apiv1.TiCDCSinkEndpoint_DNSRecord {
	ret := make([]*apiv1.TiCDCSinkEndpoint_DNSRecord, 0, len(records))
	for _, record := range records {
		ret = append(ret, &apiv1.TiCDCSinkEndpoint_DNSRecord{
			AvailabilityZone: record.AvailabilityZone,
			RecordName:       record.RecordName,
		})
	}
	return ret
}

func convertSinkEndpointPhase(phase regionalpb.PrivateLinkEndpoint_Status) apiv1.TiCDCSinkEndpoint_Phase {
	switch phase {
	case regionalpb.PrivateLinkEndpoint_AVAILABLE:
		return apiv1.TiCDCSinkEndpoint_AVAILABLE
	case regionalpb.PrivateLinkEndpoint_CREATING:
		return apiv1.TiCDCSinkEndpoint_CREATING
	case regionalpb.PrivateLinkEndpoint_PENDING:
		return apiv1.TiCDCSinkEndpoint_PENDING_ACCEPTANCE
	case regionalpb.PrivateLinkEndpoint_DELETING:
		return apiv1.TiCDCSinkEndpoint_DELETING
	case regionalpb.PrivateLinkEndpoint_REJECTED:
		return apiv1.TiCDCSinkEndpoint_REJECTED
	case regionalpb.PrivateLinkEndpoint_FAILED:
		return apiv1.TiCDCSinkEndpoint_FAILED
	default:
		return apiv1.TiCDCSinkEndpoint_PHASE_UNSPECIFIED
	}
}