package service

import (
	"context"
	"time"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"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"

	cloudv1 "github.com/tidbcloud/ticdc-service-ng/api/v1"
	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy"
	"github.com/tidbcloud/ticdc-service-ng/pkg/config"
	"github.com/tidbcloud/ticdc-service-ng/pkg/consts"
	"github.com/tidbcloud/ticdc-service-ng/pkg/regionalservice"
)

type cdcServer struct {
	apiv1.UnimplementedTiCDCClusterServiceServer

	kubeClient  client.Client
	cfg         *config.Config
	proxyModel  clusterproxy.Model
	regionalCli regionalservice.Client
}

func NewTiCDCClusterServiceServer(cli client.Client,
	cfg *config.Config,
	proxyModel clusterproxy.Model,
	regionalCli regionalservice.Client) apiv1.TiCDCClusterServiceServer {
	s := &cdcServer{
		kubeClient:  cli,
		cfg:         cfg,
		proxyModel:  proxyModel,
		regionalCli: regionalCli,
	}
	return s
}

// CreateTiCDCCluster saves the TiCDCCluster to k8s CR
func (c cdcServer) CreateTiCDCCluster(ctx context.Context, req *apiv1.CreateTiCDCClusterRequest) (*apiv1.TiCDCCluster, error) {
	sharedPoolID := req.TicdcCluster.SharedPoolId
	logger := tilog.WithContextL(ctx).With(zap.String("sharedPool", req.TicdcCluster.SharedPoolId))
	sharedPool, err := c.regionalCli.GetSharedPoolByID(ctx, req.TicdcCluster.SharedPoolId)
	if err != nil || sharedPool == nil {
		logger.Error("Failed to get sharedPool", zap.Any("sharedPool", sharedPool), zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "sharedPool %s not found", sharedPoolID)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get sharedPool %s: %s", sharedPoolID, err)
	}
	dataPlane, err := c.regionalCli.GetDataPlaneByID(ctx, sharedPool.DataplaneName)
	if err != nil {
		logger.Error("Failed to get dataplane for sharedPool",
			zap.String("dataplane", sharedPool.DataplaneName),
			zap.String("sharedPoolID", sharedPoolID), zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get dataplane %s: %s", sharedPool.DataplaneName, err)
	}

	cdc := cloudv1.TiCDC{
		ObjectMeta: metav1.ObjectMeta{
			Name:      req.TicdcClusterId,
			Namespace: c.cfg.Namespace,
			Labels: buildLabelsForCreatingTiCDC(c.cfg.Provider, c.cfg.Region,
				sharedPoolID),
		},
		Spec: cloudv1.TiCDCSpec{
			SharedPoolID: sharedPoolID,
			Replicas:     int(req.TicdcCluster.NodeSpec.Count),
			NodeSpec: cloudv1.NodeSpec{
				CpuCore:           req.TicdcCluster.NodeSpec.CpuCore,
				MemoryGi:          req.TicdcCluster.NodeSpec.MemoryGi,
				KubernetesVersion: dataPlane.Kubernetes.Version,
			},
			Storage: cloudv1.StorageSpec{
				Type:          req.TicdcCluster.Storage.Type,
				SizeGi:        req.TicdcCluster.Storage.SizeGi,
				Iops:          req.TicdcCluster.Storage.Iops,
				ThroughputMbs: req.TicdcCluster.Storage.ThroughputMbs,
			},
			Configuration: &cloudv1.TiCDCConfiguration{
				BaseImage: req.TicdcCluster.Configuration.BaseImage,
				Tag:       req.TicdcCluster.Configuration.Tag,
				Config:    req.TicdcCluster.Configuration.Config,
				Version:   req.TicdcCluster.Configuration.Version,
			},
		},
	}

	// Save the TiCDCCluster object to Kubernetes
	if err := c.kubeClient.Create(ctx, &cdc); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "TiCDC cluster %s already exists", req.TicdcClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to create TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}
	// Build the response object
	return buildTiCDCClusterResponse(cdc), nil
}

func (c cdcServer) ListTiCDCClusters(ctx context.Context, req *apiv1.ListTiCDCClustersRequest) (*apiv1.ListTiCDCClustersResponse, error) {
	// Create a list to hold the TiCDC clusters
	list := &cloudv1.TiCDCList{}

	// Define list options to filter by namespace
	listOptions := []client.ListOption{
		client.InNamespace(c.cfg.Namespace),
		client.MatchingLabelsSelector{
			Selector: buildLabelSetForQuery(req.TidbClusterId).AsSelector(),
		},
	}

	// List the TiCDC clusters from Kubernetes
	if err := c.kubeClient.List(ctx, list, listOptions...); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to list TiCDC clusters: %s", err)
	}

	// Build the response
	clusters := make([]*apiv1.TiCDCCluster, 0, len(list.Items))
	for _, item := range list.Items {
		if len(req.TicdcClusterId) > 0 &&
			!matchTiCDCClusterID(req.TicdcClusterId, item.Name) {
			continue
		}
		clusters = append(clusters, buildTiCDCClusterResponse(item))
	}
	return &apiv1.ListTiCDCClustersResponse{TicdcClusters: clusters}, nil
}

func (c cdcServer) GetTiCDCCluster(ctx context.Context, req *apiv1.GetTiCDCClusterRequest) (*apiv1.TiCDCCluster, error) {
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}
	// Build the response object
	return buildTiCDCClusterResponse(*cdc), nil
}

func (c cdcServer) ScaleTiCDCCluster(ctx context.Context, req *apiv1.ScaleTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}

	// Update the number of replicas
	cdc.Spec.Replicas = int(req.NodeSpec.Count)
	cdc.Spec.NodeSpec = cloudv1.NodeSpec{
		CpuCore:  req.NodeSpec.CpuCore,
		MemoryGi: req.NodeSpec.MemoryGi,
	}
	cdc.Spec.Storage = cloudv1.StorageSpec{
		Type:          req.Storage.Type,
		SizeGi:        req.Storage.SizeGi,
		Iops:          req.Storage.Iops,
		ThroughputMbs: req.Storage.ThroughputMbs,
	}

	// Update the TiCDC cluster in Kubernetes
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		logger.Error("Failed to scale TiCDC cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to scale TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}

	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateTiCDCClusterConfig(ctx context.Context, req *apiv1.UpdateTiCDCClusterConfigRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}
	// Update the number of replicas
	cdc.Spec.Configuration = &cloudv1.TiCDCConfiguration{
		BaseImage: req.Configuration.BaseImage,
		Tag:       req.Configuration.Tag,
		Config:    req.Configuration.Config,
		Version:   req.Configuration.Version,
	}

	// Update the TiCDC cluster in Kubernetes
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to scale TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}
	logger.Info("TiCDC cluster configuration updated")
	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateTiCDCClusterStorage(ctx context.Context, req *apiv1.UpdateTiCDCClusterStorageRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		return nil, err
	}
	// Update the storage
	cdc.Spec.Storage = cloudv1.StorageSpec{
		Type:          req.Storage.Type,
		SizeGi:        req.Storage.SizeGi,
		Iops:          req.Storage.Iops,
		ThroughputMbs: req.Storage.ThroughputMbs,
	}

	// Update the TiCDC cluster in Kubernetes
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to update TiCDC cluster storage %s: %s", req.TicdcClusterId, err)
	}
	logger.Info("TiCDC cluster storage updated")
	return &emptypb.Empty{}, nil
}

func (c cdcServer) DeleteTiCDCCluster(ctx context.Context, request *apiv1.DeleteTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, request.TicdcClusterId)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return nil, err
	}
	if err := c.kubeClient.Delete(ctx, cdc); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Warn("ticdc cluster not found")
			return &emptypb.Empty{}, nil
		}
		logger.Error("Failed to delete cdc cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to delete cdc cluster %s: %s", request.TicdcClusterId, err)
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) RestartTiCDCCluster(ctx context.Context, req *apiv1.RestartTiCDCClusterRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	cdc, err := c.getTiCDCByID(ctx, req.TicdcClusterId)
	if err != nil {
		logger.Error("Failed to get cdc cluster", zap.Error(err))
		return nil, err
	}

	if len(cdc.Annotations) == 0 {
		cdc.Annotations = make(map[string]string)
	}
	cdc.Annotations[consts.RestartTiCDCAnnotation] = time.Now().UTC().Format(time.RFC3339)
	if err := c.kubeClient.Update(ctx, cdc); err != nil {
		logger.Error("Failed to restart TiCDC cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to restart TiCDC cluster %s: %s", req.TicdcClusterId, err)
	}

	return &emptypb.Empty{}, nil
}

func (c cdcServer) GetTiCDCClusterRole(ctx context.Context, req *apiv1.GetTiCDCClusterRoleRequest) (*apiv1.TiCDCClusterRole, error) {
	tidbClusterRole, err := c.regionalCli.GetTiDBClusterRole(ctx, req.TidbClusterId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "Failed to get cluster role: %s", err)
	}
	// The iam information is unified in a data plane, so just reuse the iam information of the tidb cluster.
	return &apiv1.TiCDCClusterRole{
		CloudIamIdentity: tidbClusterRole.NodeRole,
		CloudAccountId:   tidbClusterRole.CloudIamIdentity,
		AwsExternalId:    tidbClusterRole.ExternalId,
	}, nil
}

func (c cdcServer) CreateChangefeed(ctx context.Context, req *apiv1.CreateChangefeedRequest) (*apiv1.Changefeed, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	var (
		cf  *apiv1.Changefeed
		err error
	)
	if err = c.doInProxy(ctx, req.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		req.Changefeed.ChangefeedId = req.ChangefeedId
		cf, err = proxy.ChangefeedClient().CreateChangefeedV2(ctx, req.TicdcClusterId,
			req.TidbClusterId, req.Changefeed)
		return err
	}); err != nil {
		logger.Error("Failed to create changefeed", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to create changefeed: %s", err)
	}
	return cf, nil
}

func (c cdcServer) ListChangefeeds(ctx context.Context, request *apiv1.ListChangefeedsRequest) (*apiv1.ListChangefeedsResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	var (
		cfList *apiv1.ListChangefeedsResponse
		err    error
	)
	if err = c.doInProxy(ctx, request.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		cfList, err = proxy.ChangefeedClient().GetChangefeedListV2(ctx, request.TicdcClusterId,
			request.TidbClusterId)
		return err
	}); err != nil {
		logger.Error("Failed to list changefeed", zap.Error(err))
		return nil, err
	}
	return cfList, nil
}

func (c cdcServer) GetChangefeed(ctx context.Context, request *apiv1.GetChangefeedRequest) (*apiv1.Changefeed, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	var (
		cf  *apiv1.Changefeed
		err error
	)
	if err := c.doInProxy(ctx, request.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		cf, err = proxy.ChangefeedClient().GetChangefeedV2(ctx, request.TicdcClusterId,
			request.TidbClusterId, request.ChangefeedId)
		return err
	}); err != nil {
		logger.Error("Failed to get changefeed", zap.Error(err))
		return nil, err
	}
	return cf, err
}

func (c cdcServer) PauseChangefeed(ctx context.Context, req *apiv1.PauseChangefeedRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	if err := c.doInProxy(ctx, req.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		return proxy.ChangefeedClient().PauseChangefeedV2(ctx, req.TicdcClusterId,
			req.TidbClusterId, req.ChangefeedId)
	}); err != nil {
		logger.Error("Failed to pause changefeed", zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) ResumeChangefeed(ctx context.Context, req *apiv1.ResumeChangefeedRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	if err := c.doInProxy(ctx, req.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		return proxy.ChangefeedClient().ResumeChangefeedV2(ctx, req.TicdcClusterId,
			req.TidbClusterId, req.ChangefeedId, req.Config)
	}); err != nil {
		logger.Error("Failed to pause changefeed", zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) UpdateChangefeed(ctx context.Context, request *apiv1.UpdateChangefeedRequest) (*apiv1.Changefeed, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", request.TicdcClusterId))
	var (
		retCf *apiv1.Changefeed
		err   error
	)
	if err = c.doInProxy(ctx, request.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		retCf, err = proxy.ChangefeedClient().UpdateChangefeedV2(ctx, request.TicdcClusterId,
			request.TidbClusterId, request.ChangefeedId, request.Changefeed)
		return err
	}); err != nil {
		logger.Error("Failed to create changefeed", zap.Error(err))
		return nil, err
	}
	return retCf, nil
}

func (c cdcServer) DeleteChangefeed(ctx context.Context, req *apiv1.DeleteChangefeedRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TicdcClusterId))
	if err := c.doInProxy(ctx, req.TidbClusterId, func(proxy clusterproxy.ClusterProxy) error {
		return proxy.ChangefeedClient().DeleteChangefeedV2(ctx, req.TicdcClusterId,
			req.TidbClusterId, req.ChangefeedId)
	}); err != nil {
		logger.Error("Failed to delete changefeed", zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c cdcServer) getTiCDCByID(ctx context.Context, id string) (*cloudv1.TiCDC, error) {
	// Create a TiCDC object to hold the retrieved data
	cdc := &cloudv1.TiCDC{}
	// Define the key to identify the TiCDC cluster
	key := client.ObjectKey{
		Namespace: c.cfg.Namespace,
		Name:      id,
	}
	// Get the TiCDC cluster from Kubernetes
	if err := c.kubeClient.Get(ctx, key, cdc); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "TiCDC cluster %s not found", id)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get TiCDC cluster %s: %s", id, err)
	}
	return cdc, nil
}

func (c cdcServer) doInProxy(
	ctx context.Context,
	tidbClusterID string,
	f func(proxy clusterproxy.ClusterProxy) error) error {
	logger := tilog.WithContextL(ctx)
	proxy, err := c.proxyModel.GetClusterProxy(ctx, tidbClusterID, clusterproxy.DefaultClusterProxyTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return err
	}
	return f(proxy)
}

func buildTiCDCClusterResponse(item cloudv1.TiCDC) *apiv1.TiCDCCluster {
	var configuration *apiv1.TiCDCCluster_Configuration
	if item.Spec.Configuration != nil {
		configuration = &apiv1.TiCDCCluster_Configuration{
			BaseImage: item.Spec.Configuration.BaseImage,
			Tag:       item.Spec.Configuration.Tag,
			Version:   item.Spec.Configuration.Version,
			Config:    item.Spec.Configuration.Config,
		}
	}

	cluster := &apiv1.TiCDCCluster{
		Id:           item.Name,
		SharedPoolId: item.Spec.SharedPoolID,
		NodeSpec: &apiv1.NodeInfo{
			Count:             uint32(item.Spec.Replicas),
			CpuCore:           item.Spec.NodeSpec.CpuCore,
			MemoryGi:          item.Spec.NodeSpec.MemoryGi,
			KubernetesVersion: item.Spec.NodeSpec.KubernetesVersion,
		},
		Storage: &apiv1.TiCDCCluster_Storage{
			Type:          item.Spec.Storage.Type,
			SizeGi:        item.Spec.Storage.SizeGi,
			Iops:          item.Spec.Storage.Iops,
			ThroughputMbs: item.Spec.Storage.ThroughputMbs,
		},
		Configuration: configuration,
		Phase:         mapTiCDCClusterPhase(item.Status.Phase),
	}
	if item.Status.NodeInfo != nil {
		cluster.NodeStatus = &apiv1.NodeInfo{
			Count:             uint32(item.Status.NodeInfo.Replicas),
			CpuCore:           item.Status.NodeInfo.CpuCore,
			MemoryGi:          item.Status.NodeInfo.MemoryGi,
			KubernetesVersion: item.Status.NodeInfo.KubernetesVersion,
		}
	}
	return cluster
}

func matchTiCDCClusterID(filters []string, ticdcClusterID string) bool {
	for _, f := range filters {
		if f == ticdcClusterID {
			return true
		}
	}
	return false
}

var cdcClusterPhaseMap = map[cloudv1.TiCDCPhase]apiv1.TiCDCCluster_Phase{
	cloudv1.TiCDCPhaseCreating:    apiv1.TiCDCCluster_CREATING,
	cloudv1.TiCDCPhaseRunning:     apiv1.TiCDCCluster_RUNNING,
	cloudv1.TiCDCPhaseUnavailable: apiv1.TiCDCCluster_UNAVAILABLE,
	cloudv1.TiCDCPhaseScaling:     apiv1.TiCDCCluster_SCALING,
	cloudv1.TiCDCPhaseUpgrading:   apiv1.TiCDCCluster_UPGRADING,
	cloudv1.TiCDCPhaseDeleting:    apiv1.TiCDCCluster_DELETING,
}

func mapTiCDCClusterPhase(phase cloudv1.TiCDCPhase) apiv1.TiCDCCluster_Phase {
	if v, ok := cdcClusterPhaseMap[phase]; ok {
		return v
	}
	return apiv1.TiCDCCluster_PHASE_UNSPECIFIED
}