package event_svc

import (
	"context"
	"fmt"
	"slices"
	"strconv"

	core_event "github.com/tidbcloud/core-svc/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"k8s.io/utils/ptr"
)

type BranchSyncEventService struct {
	branchRepo  repo.BranchRepo
	infraFacade facade.InfraFacade
	branchSvc   service.BranchService
	mgmtFacade  facade.MgmtSvcFacade
}

func NewBranchSyncEventService(
	branchRepo repo.BranchRepo,
	infraFacade facade.InfraFacade,
	branchSvc service.BranchService,
	mgmtfacade facade.MgmtSvcFacade,
) *BranchSyncEventService {
	return &BranchSyncEventService{
		branchRepo:  branchRepo,
		infraFacade: infraFacade,
		branchSvc:   branchSvc,
		mgmtFacade:  mgmtfacade,
	}
}

const (
	ClusterEventTypeUpdatedPublicEndpoint = core_event.ClusterEventType("ClusterPublicEndpointUpdated")
	ClusterEventTypeUpdateCapacity        = core_event.ClusterEventType("ClusterCapacityUpdated")
)

func (s *BranchSyncEventService) HandleClusterEvent(ctx context.Context, event *core_event.ClusterEvent) (err error) {
	ctx, span := Tracer.Start(ctx, "branchSyncSvc.HandleClusterEvent")
	logger := utils.GetEventBusLoggerFromContext(logid.WithLogid(ctx, logid.GenLogid())).
		With(zap.Uint64("cluster_id", event.ClusterID)).
		With(zap.String("event_type", string(event.EventType))).
		With(zap.String("svc", "branchSyncSvc"))

	logger.Info("start to handle cluster event")
	defer func() {
		tracing.EndSpan(span, err)
		if err != nil {
			logger.Error("failed to handle cluster event", zap.Error(err))
		} else {
			logger.Info("succeed to handle cluster event")
			return
		}

		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		eventID := ctx.Value(config.CtxEventIDKey)
		eventIDString := "unknown"
		if eventID != nil {
			if id, ok := eventID.(string); ok {
				eventIDString = id
			}
		}

		metric.BranchSyncHandlerCounter.WithLabelValues(eventIDString, strconv.FormatUint(event.ClusterID, 10), string(event.EventType), status).
			Inc()
	}()

	if event.EventType == core_event.ClusterEventTypeDeleted {
		return s.handleDeleted(ctx, logger, event.ClusterID)
	}
	if event.EventType == ClusterEventTypeUpdatedPublicEndpoint {
		return s.handlePublicEndpointEvent(ctx, logger, event.ClusterID)
	}
	if event.EventType == ClusterEventTypeUpdateCapacity {
		return s.handleClusterCapacityEvent(ctx, logger, event.ClusterID)
	}

	return nil
}

func (s *BranchSyncEventService) handleDeleted(ctx context.Context, logger *zap.Logger, clusterID uint64) error {
	logger.Info("start to handle cluster deleted event")
	err := s.branchSvc.DeleteBranchByClusterId(ctx, strconv.FormatUint(clusterID, 10), constant.UnknownAuthorID)
	if err != nil {
		logger.Error("failed to delete child branches", zap.Error(err))
		return err
	}
	return nil
}

func (s *BranchSyncEventService) handlePublicEndpointEvent(
	ctx context.Context,
	logger *zap.Logger,
	clusterID uint64,
) error {
	logger.Info("start to handle cluster update public endpoint event")
	branches, bizErr := s.branchRepo.ListBranchesByMainBranchID(ctx, strconv.FormatUint(clusterID, 10), nil)
	if bizErr != nil {
		logger.Error("list branches by root branch id failed", zap.Error(bizErr))
		return bizErr
	}
	if len(branches) == 0 {
		logger.Info("no child branches")
		return nil
	}

	cluster, err := s.mgmtFacade.GetCluster(ctx, clusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		if err.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Info("cluster not found, skip")
			return nil
		}
		logger.Error("get cluster failed", zap.Error(err))
		return err
	}

	for _, branch := range branches {
		branLog := logger.With(zap.String("child_branch_id", branch.ID))
		br, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
		if err != nil {
			branLog.Error("get branch CR failed", zap.Error(err))
			return err
		}

		prev := br.Spec.SqlEndpoints.Public.IsEnabled()
		clusterAuthorizedNetworks := service.ConvertRegionalToInfraAPIAuthorizedNetworks(
			cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().GetAuthorizedNetwork(),
		)
		if !cluster.GetServerlessCluster().
			GetCluster().
			GetEndpoints().
			GetPublic().
			Disabled != br.Spec.SqlEndpoints.Public.IsEnabled() ||
			!slices.Equal(clusterAuthorizedNetworks, br.Spec.SqlEndpoints.Public.AuthorizedNetworks) {
			br.Spec.SqlEndpoints.Public.Enabled = ptr.To(
				!cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().Disabled,
			)
			br.Spec.SqlEndpoints.Public.AuthorizedNetworks = clusterAuthorizedNetworks
			err := s.infraFacade.UpdateBranch(ctx, br)
			if err != nil {
				branLog.Error("failed to update public endpoint of child branch", zap.Error(err))
				return err
			}
		}

		branLog.Info(
			"succeed to update public endpoint of child branch",
			zap.Bool("previous_enabled", prev),
			zap.Bool(
				"current_enabled",
				!cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().Disabled,
			),
			zap.String("current_authorized_networks", fmt.Sprintf("%v", clusterAuthorizedNetworks)),
		)
	}

	logger.Info("succeed to update public endpoint of all child branches")
	return nil
}

func (s *BranchSyncEventService) handleClusterCapacityEvent(
	ctx context.Context,
	logger *zap.Logger,
	clusterID uint64,
) error {
	logger.Info("start to handle cluster update capacity event")
	branches, bizErr := s.branchRepo.ListBranchesByMainBranchID(ctx, strconv.FormatUint(clusterID, 10), nil)
	if bizErr != nil {
		logger.Error("list branches by root branch id failed", zap.Error(bizErr))
		return bizErr
	}
	if len(branches) == 0 {
		logger.Info("no child branches")
		return nil
	}

	cluster, err := s.mgmtFacade.GetCluster(ctx, clusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		if err.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Info("cluster not found, skip")
			return nil
		}
		logger.Error("get cluster failed", zap.Error(err))
		return err
	}

	for _, branch := range branches {
		branLog := logger.With(zap.String("child_branch_id", branch.ID))
		br, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
		if err != nil {
			branLog.Error("get branch CR failed", zap.Error(err))
			return err
		}

		if br.Spec.AutoScaling == nil {
			br.Spec.AutoScaling = &cloudv1.AutoScaling{}
		}

		prev := *br.Spec.AutoScaling
		now := cloudv1.AutoScaling{
			MinRcu: cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu(),
			MaxRcu: cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu(),
		}
		if prev != now {
			br.Spec.AutoScaling = &now
			// hardcode the pool now since at this time the parent cluster pool may not be updated yet
			if br.Annotations[constant.AnnoTidbPoolSelector] == constant.TidbPoolNameFree {
				br.Annotations[constant.AnnoTidbPoolSelector] = constant.TidbPoolNameVip
			}
			err := s.infraFacade.UpdateBranch(ctx, br)
			if err != nil {
				branLog.Error("failed to update capcity of child branch", zap.Error(err))
				return err
			}
		}

		branLog.Info(
			"succeed to update capcity of child branch",
			zap.Any("previous_capcity", prev),
			zap.Any("current_capcity", now),
		)
	}

	logger.Info("succeed to update capcity of all child branches")
	return nil
}