package service

import (
	"context"
	"fmt"
	"strings"

	asappsv1 "github.com/pingcap/advanced-statefulset/client/apis/apps/v1"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"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"

	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-agent/v1"
)

type cdcAgentService struct {
	apiv1.UnimplementedCdcAgentServiceServer

	dataplaneCli client.Client
}

func NewCdcAgentService(dataplaneCli client.Client) *cdcAgentService {
	return &cdcAgentService{
		dataplaneCli: dataplaneCli,
	}
}

// PauseTC pauses a single CDC TC within a TiDB cluster
func (s *cdcAgentService) PauseTC(ctx context.Context, req *apiv1.CdcPauseTCReq) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("tidb_cluster_id", req.TidbClusterId),
		zap.String("tc_name", req.TcName),
	)
	logger.Info("PauseTC called")

	tc, err := validateCdcTc(ctx, s.dataplaneCli, req.TidbClusterId, req.TcName)
	if err != nil {
		logger.Error("PauseTC validation failed", zap.Error(err))
		return nil, err
	}

	// check if tc is already paused
	if tc.Spec.Paused {
		logger.Info("TC is already paused, no action needed")
		return &emptypb.Empty{}, nil
	}

	logger.Info("Setting TC to paused state")
	tc.Spec.Paused = true
	if err := s.dataplaneCli.Update(ctx, tc); err != nil {
		logger.Error("Failed to pause ticdc TC", zap.Error(err))
		return nil, status.Error(codes.Internal, fmt.Sprintf("failed to pause ticdc TC: %s", err.Error()))
	}

	logger.Info("PauseTC completed successfully")
	return &emptypb.Empty{}, nil
}

// ResumeTC resumes a single CDC TC within a TiDB cluster
func (s *cdcAgentService) ResumeTC(ctx context.Context, req *apiv1.CdcResumeTCReq) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("tidb_cluster_id", req.TidbClusterId),
		zap.String("tc_name", req.TcName),
	)
	logger.Info("ResumeTC called")

	tc, err := validateCdcTc(ctx, s.dataplaneCli, req.TidbClusterId, req.TcName)
	if err != nil {
		logger.Error("ResumeTC validation failed", zap.Error(err))
		return nil, err
	}

	// check if tc is already resumed
	if !tc.Spec.Paused {
		logger.Info("TC is already resumed, no action needed")
		return &emptypb.Empty{}, nil
	}

	logger.Info("Setting TC to resumed state")
	tc.Spec.Paused = false
	if err := s.dataplaneCli.Update(ctx, tc); err != nil {
		logger.Error("Failed to resume ticdc TC", zap.Error(err))
		return nil, status.Error(codes.Internal, fmt.Sprintf("failed to resume ticdc TC: %s", err.Error()))
	}

	logger.Info("ResumeTC completed successfully")
	return &emptypb.Empty{}, nil
}

// DeleteAstsAndResumeTC deletes ASTS for CDC component and optionally resumes TC
func (s *cdcAgentService) DeleteAstsAndResumeTC(ctx context.Context, req *apiv1.CdcDeleteAstsAndResumeTCReq) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(
		zap.String("tidb_cluster_id", req.TidbClusterId),
		zap.String("tc_name", req.TcName),
		zap.Bool("only_delete_asts", req.OnlyDeleteAsts),
	)
	logger.Info("DeleteAstsAndResumeTC called")

	// validation
	logger.Info("Validating CDC TC")
	tc, err := validateCdcTc(ctx, s.dataplaneCli, req.TidbClusterId, req.TcName)
	if err != nil {
		logger.Error("DeleteAstsAndResumeTC validation failed", zap.Error(err))
		return nil, err
	}

	// get asts
	logger.Info("Getting ASTS for CDC component",
		zap.String("component", "ticdc"),
		zap.String("instance", req.TcName))
	labels := map[string]string{
		K8sComponentLabelKey: "ticdc",
		K8sInstanceLabelKey:  req.TcName,
	}
	namespace := buildNamespace(req.TidbClusterId)
	asts := &asappsv1.StatefulSetList{}
	if err := s.dataplaneCli.List(ctx, asts, client.InNamespace(namespace), client.MatchingLabels(labels)); err != nil {
		logger.Error("Failed to list ASTS", zap.Error(err))
		return nil, status.Error(codes.Internal, "failed to list asts: "+err.Error())
	}

	logger.Info("Found ASTS to delete", zap.Int("count", len(asts.Items)))

	// delete asts
	deletedCount := 0
	for i := range asts.Items {
		ast := asts.Items[i]
		logger.Info("Deleting ASTS",
			zap.Int("current", i+1),
			zap.Int("total", len(asts.Items)),
			zap.String("namespace", ast.Namespace),
			zap.String("name", ast.Name))

		if err := s.dataplaneCli.Delete(ctx, &ast, client.PropagationPolicy(metav1.DeletePropagationForeground)); err != nil {
			logger.Error("Failed to delete ASTS",
				zap.String("namespace", ast.Namespace),
				zap.String("name", ast.Name),
				zap.Error(err))
			return nil, status.Error(codes.Internal, "failed to delete ast: "+err.Error())
		}
		deletedCount++
		logger.Info("Successfully deleted ASTS",
			zap.Int("current", i+1),
			zap.Int("total", len(asts.Items)),
			zap.String("namespace", ast.Namespace),
			zap.String("name", ast.Name))
	}

	if req.OnlyDeleteAsts {
		logger.Info("OnlyDeleteAsts is true, skipping TC resume", zap.Int("deleted_asts", deletedCount))
		return &emptypb.Empty{}, nil
	}

	// resume TC
	if !tc.Spec.Paused {
		logger.Info("TC is already resumed, no action needed")
		return &emptypb.Empty{}, nil
	}

	logger.Info("Resuming TC after ASTS deletion")
	tc.Spec.Paused = false
	if err := s.dataplaneCli.Update(ctx, tc); err != nil {
		logger.Error("Failed to resume ticdc TC", zap.Error(err))
		return nil, status.Error(codes.Internal, fmt.Sprintf("failed to resume ticdcTC: %s", err.Error()))
	}

	logger.Info("DeleteAstsAndResumeTC completed successfully",
		zap.Int("deleted_asts", deletedCount))
	return &emptypb.Empty{}, nil
}

func validateCdcTc(ctx context.Context, dataplaneCli client.Client, clusterId, ticdcTCName string) (*v1alpha1.TidbCluster, error) {
	if clusterId == "" {
		return nil, status.Error(codes.InvalidArgument, "tidb_cluster_id is required")
	}
	if err := validateCdcTcName(ticdcTCName); err != nil {
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}

	tc := &v1alpha1.TidbCluster{}
	if err := dataplaneCli.Get(ctx, client.ObjectKey{
		Namespace: buildNamespace(clusterId),
		Name:      ticdcTCName,
	}, tc); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, status.Error(codes.NotFound, "ticdc TC not found")
		}
		return nil, status.Error(codes.Internal, "failed to get ticdc TC: "+err.Error())
	}
	return tc, nil
}

func validateCdcTcName(tcName string) error {
	if tcName == "" {
		return status.Error(codes.InvalidArgument, "tc_name should not be empty")
	}
	if !strings.HasPrefix(tcName, "ticdc-") && !strings.HasPrefix(tcName, "changefeed-") {
		return status.Error(codes.InvalidArgument, "tc_name should be prefixed with 'ticdc-' or 'changefeed-'")
	}
	return nil
}