package connprofile

import (
	"context"
	"net"
	"strings"

	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/core_ctx/domain/facade"
	"github.com/tidbcloud/dataflow-service/internal/core_ctx/infra/controller"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	entity2 "github.com/tidbcloud/dataflow-service/app/repo/entity"
	agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/common"
	commondm "github.com/tidbcloud/dataflow-service/internal/common/dm"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/internal/handler/agent"
	"github.com/tidbcloud/dataflow-service/internal/model/connprofile"
	"github.com/tidbcloud/dataflow-service/internal/model/dm"
	"github.com/tidbcloud/dataflow-service/internal/model/feature"
	"github.com/tidbcloud/dataflow-service/internal/service"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/pkgv2/logid"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"gorm.io/gorm"
)

type connProfileService struct {
	dataflow.UnimplementedConnProfileServiceServer
	service.OperationsService
	model connprofile.Model
}

func NewService(operationsService service.OperationsService, model connprofile.Model) dataflow.ConnProfileServiceServer {
	return &connProfileService{
		OperationsService: operationsService,
		model:             model,
	}
}

func (s *connProfileService) userCtx(ctx context.Context) connprofile.UserContext {
	user := s.CurrentUser(ctx)
	return s.model.UserContext(ctx, user)
}

// todo: quite duplicate with dm/service, merge them later.
func (s *connProfileService) checkUser(ctx context.Context, projectID, clusterID uint64) (*connprofile.UserContext, *core.Cluster, error) {
	userCtx := s.userCtx(ctx)
	logger := userCtx.Log()
	if err := userCtx.CheckProjectPermission(entity2.ID(projectID)); err != nil {
		return nil, nil, err
	}
	cluster, err := grpcli.GetClusterWithDataplaneByID(ctx, clusterID)
	if err != nil {
		logger.Error("get target cluster failed", zap.Error(err))
		// todo: difference between not-found and internal error
		return nil, nil, errcode.NewInvalidInputErr(errors.New("Failed to get cluster information. Please check whether the cluster is cleared or deleted."))
	}
	if cluster.ProjectId != projectID {
		return nil, nil, errcode.NewInvalidInputErr(errors.New("project id mismatch with cluster's project id"))
	}
	return &userCtx, cluster, nil
}

func (s *connProfileService) commonCheck(ctx context.Context, projectID, clusterID uint64) (*connprofile.UserContext, *core.Cluster, error) {
	userCtx, cluster, err := s.checkUserAndCluster(ctx, projectID, clusterID)
	if err != nil {
		return nil, nil, err
	}
	if !feature.IsFeatureEnabledByCluster(feature.DM, cluster) {
		return nil, nil, errcode.NewInvalidInputErr(errors.New("Data migration is not enabled on this cluster, please check whether this cluster meets the prerequisites in the doc."))
	}
	return userCtx, cluster, nil
}

func (s *connProfileService) checkUserAndCluster(ctx context.Context, projectID, clusterID uint64) (*connprofile.UserContext, *core.Cluster, error) {
	userCtx, cluster, err := s.checkUser(ctx, projectID, clusterID)
	if err != nil {
		return nil, nil, err
	}
	if !bizutils.ClusterAvailable(cluster) {
		return nil, nil, errcode.NewInvalidInputErr(errors.New("cluster is not in normal or scaling status"))
	}
	return userCtx, cluster, nil
}

func (s *connProfileService) CreateConnProfile(ctx context.Context, req *dataflow.CreateConnProfileReq) (*dataflow.CreateConnProfileResp, error) {
	var err error
	resp := &dataflow.CreateConnProfileResp{}
	userCtx, cluster, err := s.commonCheck(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	logger := userCtx.Log()
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster, network, agent client failed")
	}
	var (
		endpointServiceName, host, host2Connect string
		isSameRegion                            bool
		err2                                    error
	)
	switch entity.ConnMethod(req.ConnMethod) {
	case entity.ConnMethodPrivateLink:
		if req.EndpointServiceName == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Endpoint service name should not be empty when using private link"))
		}
		endpointServiceName = strings.TrimSpace(*req.EndpointServiceName)
		privateLink, err2 := connprofile.GetAvailablePrivateLink(cluster.GetDataPlaneInfo().GetK8SNamespace(), cluster.GetDataPlaneInfo().GetNetworkName(), endpointServiceName)
		if err2 != nil {
			logger.Error("get private link info failed", zap.Error(err2))
			return nil, errcode.NewInternalErr(errors.New("Failed to get available private link endpoint."))
		}
		host2Connect = privateLink.RegionalDNS
	case entity.ConnMethodDirect:
		if req.Host == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Host should not be empty when using Public IP."))
		}
		host = strings.TrimSpace(*req.Host)

		// user input host/ip should not be a vpc-peering if user choose direct connect
		// we check it since we use connect-method to do metering.
		if isSameRegion, err2 = s.checkVpcPeering(ctx, logger, cluster, agentClient, host); err2 != nil {
			if !s.isVpcPeeringNotFoundErr(err2) {
				return nil, err2
			}
			// vpc-peering is not found, that's what we expected.
		} else {
			return nil, errcode.NewInvalidInputErr(errors.New("Given host resolved to a private address of some vpc-peering, please choose VPC Peering connectivity method if you want to use it."))
		}

		host2Connect = host
	case entity.ConnMethodVPCPeering:
		if req.Host == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Host should not be empty when using VPC Peering."))
		}
		host = strings.TrimSpace(*req.Host)
		// user input host must be a valid address in some peering vpc
		if isSameRegion, err2 = s.checkVpcPeering(ctx, logger, cluster, agentClient, host); err2 != nil {
			return nil, err2
		}

		host2Connect = host
	}

	if err = utils.CheckTLSParam(req.SslCert, req.SslKey); err != nil {
		return nil, err
	}

	dbConnResp, err := agentClient.CheckDBConnectivity(ctx, &agentpb.CheckDBConnectivityReq{
		Host:      host2Connect,
		Port:      req.Port,
		User:      req.User,
		Password:  req.Password,
		SslCa:     req.SslCa,
		SslCert:   req.SslCert,
		SslKey:    req.SslKey,
		RequestId: logid.GetLogid(ctx),
	})
	if err != nil {
		// todo: all error is taken as invalid input, handle by error type
		logger.Error("failed to check connectivity on create conn profile", zap.Error(err))
		return nil, commondm.NormalizeErr(err, "failed to check connectivity on create conn profile")
	}

	var encryptedPassword string
	if err = controller.DiCall(func(encryptor facade.Encryptor) error {
		var err2 error
		encryptedPassword, err2 = encryptor.EncryptBase64(req.Password)
		return err2
	}); err != nil {
		// don't want user know we failed to encrypt
		return nil, errcode.NewInternalErr(errors.Errorf("failed to create connection profile"))
	}

	connProfile := &entity.DFConnProfile{
		TenantID:            entity.ID(cluster.TenantId),
		ProjectID:           entity.ID(cluster.ProjectId),
		ClusterID:           entity.ID(cluster.Id),
		CreatorUID:          entity.ID(userCtx.UserID()),
		Name:                req.Name,
		Type:                entity.DBType(req.Type),
		Version:             dbConnResp.DbVersion,
		Region:              req.Region,
		ConnMethod:          entity.ConnMethod(req.ConnMethod),
		EndpointServiceName: endpointServiceName,
		Host:                host,
		Port:                int32(req.Port),
		User:                req.User,
		EncryptedPassword:   encryptedPassword,
		SslCa:               req.SslCa,
		SslCert:             req.SslCert,
		SslKey:              req.SslKey,
		SameRegion:          isSameRegion,
	}

	resp.Id, err = userCtx.Create(connProfile)

	return resp, err
}

func (s *connProfileService) GetIPWhiteList(ctx context.Context, req *dataflow.GetIPWhiteListReq) (*dataflow.GetIPWhiteListResp, error) {
	_, cluster, err := s.checkUser(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	network, err := common.GetNetworkByCluster(ctx, cluster)
	if err != nil {
		return nil, err
	}
	if entity.ConnMethod(req.ConnMethod) == entity.ConnMethodVPCPeering {
		return &dataflow.GetIPWhiteListResp{Ips: []string{network.GetCidr()}}, nil
	}
	// else it's direct connection, use external addresses
	return &dataflow.GetIPWhiteListResp{Ips: network.GetExternalIPs()}, nil
}

func (s *connProfileService) CreatePrivateLinkEndpoint(ctx context.Context, req *dataflow.CreatePrivateLinkEndpointReq) (*dataflow.CreatePrivateLinkEndpointResp, error) {
	resp := &dataflow.CreatePrivateLinkEndpointResp{}
	userCtx, cluster, err := s.checkUserAndCluster(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}
	logger := userCtx.Log()

	// check whether there's existing private link endpoint with same (network, service-name)
	// which is not failed or deleted, i.e. it's in pending/pendingAcceptance/available status
	// if exists we reuse it, else we create a new one
	endpointServiceName := strings.TrimSpace(req.EndpointServiceName)
	if endpointServiceName == "" {
		switch cluster.GetInfraProviderType() {
		case infraapi.ProviderTypeEKS:
			return nil, errcode.NewInvalidInputErr(errors.New("Endpoint Service Name is required."))
		case infraapi.ProviderTypeAKS:
			return nil, errcode.NewInvalidInputErr(errors.New("Private Endpoint Resource ID is required."))
		default:
			return nil, errcode.NewInvalidInputErr(errors.New("Private link is not supported by the current provider."))
		}
	}

	plEndpoint, err := connprofile.GetNonFailedNonDeletedPLEndpoint(
		cluster.GetDataPlaneInfo().GetK8SNamespace(), cluster.GetDataPlaneInfo().GetNetworkName(), endpointServiceName)
	if err != nil && errors.Cause(err) != gorm.ErrRecordNotFound {
		return nil, errcode.NewInternalErr(errors.Errorf("failed to get private link"))
	}
	if plEndpoint != nil {
		resp.Id = uint64(plEndpoint.ID)
		return resp, nil
	}

	// note: since there's no distributed lock, we might create multiple pl-endpoint with same
	// (network, service-name) param, the pl-endpoint worker will deduplicate it.
	// note: we cannot make (network, service-name) unique in metadb table, since we
	// need to support recreation and retain history.
	// todo: use distributed lock to avoid create duplicate pl-endpoint
	resp.Id, err = userCtx.CreatePLEndpoint(cluster, endpointServiceName)
	if err != nil {
		msg := "failed to create private link endpoint"
		logger.Error(msg, zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(msg))
	}
	return resp, nil
}

func (s *connProfileService) GetPrivateLinkEndpoint(ctx context.Context, req *dataflow.GetPrivateLinkEndpointReq) (*dataflow.GetPrivateLinkEndpointResp, error) {
	resp := &dataflow.GetPrivateLinkEndpointResp{}
	_, cluster, err := s.checkUserAndCluster(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	plEndpoint, err := connprofile.GetPLEndpointByID(entity.ID(req.Id))
	if err != nil {
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewNotFoundErr(err)
		}
		return nil, errcode.NewInternalErr(errors.New("get private link endpoint failed"))
	}
	if plEndpoint.NetworkNS != cluster.GetDataPlaneInfo().GetK8SNamespace() || plEndpoint.NetworkName != cluster.GetDataPlaneInfo().GetNetworkName() {
		// we don't expose network concept, so the returned message is "mismatch with cluster"
		return nil, errcode.NewInvalidInputErr(errors.New("private link mismatch with the cluster id"))
	}
	resp.EndpointId = plEndpoint.EndpointID
	switch plEndpoint.Phase {
	case entity.PLEndpointPhaseNew, entity.PLEndpointPhasePending:
		resp.Phase = dataflow.GetPrivateLinkEndpointResp_Pending
	case entity.PLEndpointPhasePendingAcceptance:
		resp.Phase = dataflow.GetPrivateLinkEndpointResp_PendingAcceptance
	case entity.PLEndpointPhaseFailed:
		resp.Phase = dataflow.GetPrivateLinkEndpointResp_Failed
		resp.Message = plEndpoint.ErrMsg
	case entity.PLEndpointPhaseDeleted:
		// should not happen, we wait for 5m before delete failed pl-endpoints.
		return nil, errcode.NewNotFoundErr(errors.New("Failed to create private link endpoint or the created endpoint is idle for too long and deleted automatically."))
	case entity.PLEndpointPhaseAvailable:
		resp.Phase = dataflow.GetPrivateLinkEndpointResp_Available
	}
	return resp, nil
}

var vpcPeeringNotFoundErr = errors.New("Cannot find matched VPC peering, please check whether VPC peering is created on this project and check whether DNS resolution is enabled for the VPC peering if you're using host.")

func (s *connProfileService) isVpcPeeringNotFoundErr(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(err.Error(), vpcPeeringNotFoundErr.Error())
}

func (s *connProfileService) vpcPeeringAddr2IPs(ctx context.Context, agentClient agent.AgentClient, addr string) ([]net.IP, error) {
	if ip := net.ParseIP(addr); ip != nil {
		return []net.IP{ip}, nil
	}
	return commondm.VpcPeeringHost2IP(ctx, agentClient, addr)
}

// return (isSameRegion, error)
func (s *connProfileService) checkVpcPeering(ctx context.Context, logger *zap.Logger, cluster *core.Cluster, agentClient agent.AgentClient, addr string) (bool, error) {
	iPs, err2 := s.vpcPeeringAddr2IPs(ctx, agentClient, addr)
	if err2 != nil {
		logger.Error("failed to convert vpc-peering addr to ips", zap.Error(err2))
		return false, errcode.NewInvalidInputErr(errors.New("Failed to resolve the given host."))
	}
	peeringInfo, err2 := dm.GetVpcPeering(ctx, cluster, iPs)
	if err2 != nil {
		if dm.IsVpcPeeringNotFound(err2) {
			logger.Warn("vpc-peering not found", zap.Error(err2))
			return false, errcode.NewInvalidInputErr(vpcPeeringNotFoundErr)
		} else {
			logger.Error("failed to get vpc-peering info", zap.Error(err2))
			return false, errcode.NewInternalErr(errors.New("Failed to check whether input host/ip is private address of vpc peering or not, please try again later."))
		}
	}
	logger.Info("found peering info",
		zap.String("cluster-cloud-provider", cluster.CloudProviderName),
		zap.String("cluster-region", cluster.RegionName),
		zap.Uint64("peering-id", peeringInfo.Id),
		zap.String("peering-cidr", peeringInfo.Cidr),
		zap.String("tenant-region", peeringInfo.TenantRegion))
	return peeringInfo.TenantRegion == cluster.RegionName, nil
}

func (s *connProfileService) ValidateRemoteDBConnectivity(ctx context.Context, req *dataflow.ValidateRemoteDBConnectivityReq) (*dataflow.ValidateRemoteDBConnectivityResp, error) {
	resp := &dataflow.ValidateRemoteDBConnectivityResp{}
	userCtx, cluster, err := s.commonCheck(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}
	logger := userCtx.Log()
	_, agentClient, err := common.GetNetworkAgentByCluster(ctx, cluster)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster, network, agent client failed")
	}

	var host string
	switch entity.ConnMethod(req.ConnMethod) {
	case entity.ConnMethodPrivateLink:
		if req.EndpointServiceName == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Endpoint service name should not be empty when using private link"))
		}
		endpointServiceName := strings.TrimSpace(*req.EndpointServiceName)
		privateLink, err2 := connprofile.GetAvailablePrivateLink(
			cluster.GetDataPlaneInfo().GetK8SNamespace(), cluster.GetDataPlaneInfo().GetNetworkName(), endpointServiceName)
		if err2 != nil {
			logger.Error("get private link info failed", zap.Error(err2))
			return nil, errcode.NewInternalErr(errors.New("Failed to get available private link endpoint."))
		}
		host = privateLink.RegionalDNS
	case entity.ConnMethodDirect:
		if req.Host == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Host should not be empty when using Public IP."))
		}
		host = strings.TrimSpace(*req.Host)

		// // if user input an ip, the ip should not be an IP in current Network's CIDR.
		// if ip := net.ParseIP(host); ip != nil {
		// 	_, ipNet, err2 := net.ParseCIDR(network.Spec.CIDR)
		// 	if err2 != nil {
		// 		// should not happen
		// 		logger.Error("failed to parse network CIDR", zap.Error(err2))
		// 		return nil, errcode.NewInternalErr(errors.New("Invalid cluster network."))
		// 	}
		// 	if ipNet.Contains(ip) {
		// 		return nil, errcode.NewInvalidInputErr(errors.New("Given host is invalid, please input a correct host."))
		// 	}
		// }
		// user input host/ip should not be a vpc-peering if user choose direct connect
		// we check it since we use connect-method to do metering.
		if _, err2 := s.checkVpcPeering(ctx, logger, cluster, agentClient, host); err2 != nil {
			if !s.isVpcPeeringNotFoundErr(err2) {
				return nil, err2
			}
			// vpc-peering is not found, that's what we expected.
		} else {
			return nil, errcode.NewInvalidInputErr(errors.New("Given host resolved to a private address of some vpc-peering, please choose VPC Peering connectivity method if you want to use it."))
		}
	case entity.ConnMethodVPCPeering:
		if req.Host == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Host should not be empty when using VPC Peering."))
		}
		host = strings.TrimSpace(*req.Host)
		// user input host/ip must be a valid address in some peering vpc
		if _, err2 := s.checkVpcPeering(ctx, logger, cluster, agentClient, host); err2 != nil {
			return nil, err2
		}
	}

	if err = utils.CheckTLSParam(req.SslCert, req.SslKey); err != nil {
		return nil, err
	}

	dbConnResp, err := agentClient.CheckDBConnectivity(ctx, &agentpb.CheckDBConnectivityReq{
		Host:      host,
		Port:      req.Port,
		User:      req.User,
		Password:  req.Password,
		SslCa:     req.SslCa,
		SslCert:   req.SslCert,
		SslKey:    req.SslKey,
		RequestId: logid.GetLogid(ctx),
	})
	if err != nil {
		// todo: all error is taken as invalid input, handle by error type
		logger.Error("failed to validate remote connectivity", zap.Error(err))
		return nil, commondm.NormalizeErr(err, "failed to validate db connectivity")
	}
	resp.DbVersion = dbConnResp.DbVersion
	return resp, nil
}

func (s *connProfileService) ValidateClusterConnectivity(ctx context.Context, req *dataflow.ValidateClusterConnectivityReq) (*emptypb.Empty, error) {
	resp := &emptypb.Empty{}
	userCtx, cluster, err := s.commonCheck(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}
	logger := userCtx.Log()
	_, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster, network, agent client failed")
	}

	tcNameGetter := infraapi.ClusterNameGetter{
		InfraProvider:        cluster.InfraProviderType,
		InfraAPIObjNamespace: infraapi.GetObjectNamespace(req.ProjectId),
		ClusterName:          infraapi.ClusterName(req.ClusterId),
	}
	var sslCa string
	// only consider dedicated-tier, free-tier don't have client secret
	tlsConfig := config.Global().TlsConfig
	if tlsConfig.IsEnableTLS(req.OrgId, req.ProjectId) {
		tidbServerCertSecret, err := agentClient.GetTiDBServerSecret(ctx, &agentpb.GetTiDBServerSecretReq{
			ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(),
			ClusterNamespace:  tcNameGetter.ClusterNamespace(),
			ProviderType:      cluster.InfraProviderType,
		})
		if err != nil {
			logger.Error("failed to get cluster client secret", zap.Error(err))
			return nil, errcode.NewInternalErr(errors.New("failed to get cluster"))
		}
		sslCa = tidbServerCertSecret.SslCa
	}

	_, err = agentClient.CheckDBConnectivity(ctx, &agentpb.CheckDBConnectivityReq{
		Host: tcNameGetter.TiDBNamespacedServiceName(),
		// user can self-define port of tidb, but port of db-tidb(k8s service) is always 4000
		Port:      4000,
		User:      req.User,
		Password:  req.Password,
		SslCa:     sslCa,
		RequestId: logid.GetLogid(ctx),
	})
	if err != nil {
		// todo: all error is taken as invalid input, handle by error type
		logger.Error("failed to validate cluster connectivity", zap.Error(err))
		return nil, commondm.NormalizeErr(err, "failed to validate db connectivity")
	}
	return resp, nil
}